From d8d48c26a35e435709ed877a123e2499d1011757 Mon Sep 17 00:00:00 2001 From: Eric Wang Date: Sat, 24 Oct 2026 17:08:46 +0800 Subject: [PATCH] feat(llm): migrate model calls to LangChain ChatOpenAI --- .env.example | 3 +- README.md | 16 + docs/assignment6-compat.md | 5 + docs/langgraph-backend-migration-plan.md | 14 +- requirements.txt | 2 + src/agent/chat_model.py | 51 ++++ src/agent/llm.py | 99 ++++--- test/agent/test_chat_openai_adapter.py | 291 +++++++++++++++++++ test/agent/test_llm_error_mapping.py | 4 +- test/agent/test_llm_stream_tokens.py | 32 +- test/agent/test_workflow_prompts.py | 10 +- test/services/test_field_commit_and_retry.py | 38 ++- test/services/test_stream_retry_cleanup.py | 20 +- 13 files changed, 489 insertions(+), 96 deletions(-) create mode 100644 src/agent/chat_model.py create mode 100644 test/agent/test_chat_openai_adapter.py diff --git a/.env.example b/.env.example index b554f28..031824b 100644 --- a/.env.example +++ b/.env.example @@ -27,7 +27,8 @@ VOICE_CONFIG=config/voice.json # LangGraph chat backend（作业 2） # AGENT_BACKEND=langgraph # -# 命名 LLM 提供商组（均为 OpenAI 兼容接口）：LLM_PROVIDERS 声明名单， +# 命名 LLM 提供商组（通过 LangChain ChatOpenAI 统一调用 OpenAI 兼容接口）： +# LLM_PROVIDERS 声明名单， # LLM_PROVIDER 选择当前生效组；缺省或 fake = 离线脚本模型。 # 每组 LLM_<名>_BASE_URL / _API_KEY / _MODEL 必填； # 可选 _TIMEOUT_SECONDS / _MAX_RETRIES / _TEMPERATURE / _EXTRA_BODY（JSON 字符串， diff --git a/README.md b/README.md index 5c5e623..e3456f3 100644 --- a/README.md +++ b/README.md @@ -47,6 +47,22 @@ PostgreSQL 的连接池、迁移、本地 compose 与多 worker 部署见 [docs/agent-storage.md](docs/agent-storage.md)。Windows 使用 `python -m src.serve`， 它为 psycopg 选择兼容的事件循环；Linux 也可用常规 uvicorn 命令启动。 +## 模型调用 + +LangGraph 文本后端通过 `langchain-openai.ChatOpenAI` 调用 OpenAI 兼容接口。 +现有 `LLM_PROVIDERS` / `LLM_PROVIDER` 和各组环境变量继续生效；统一使用 +Chat Completions API，`EXTRA_BODY` 中的关闭推理等扩展参数照常透传。 +依赖固定为 `langchain-openai==1.1.9`、`langchain-core==1.5.1`， +与当前 LangGraph 和 OpenAI SDK 环境配套验证。 + +应用层控制重试：默认每次尝试 60 秒、最多额外重试 3 次，退避 1/2/4 秒； +底层 SDK 自动重试关闭。模型文本开始输出后不重试，提前停止消费会显式关闭 +底层 HTTP 流。兼容网关默认不请求流式 usage；非流式返回的 usage +可供 LangChain callbacks 使用。 + +模型调用继承图的 callbacks、metadata 和 tags，并标记 `llm.complete` / `llm.stream` +及 `llm_attempt`。这是追踪集成的基础，日志、存储和整轮 SSE 的观测需分别接入。 + ## HTTP 接口 - `GET /`：存活检查。 diff --git a/docs/assignment6-compat.md b/docs/assignment6-compat.md index 91c432d..706e43d 100644 --- a/docs/assignment6-compat.md +++ b/docs/assignment6-compat.md @@ -109,6 +109,11 @@ 使用与非流式相同的固定文案，不回传底层异常原文。首包前和输出途中 都使用同一分类；错误是终止事件，随后不会发送 done。已经建立的 SSE HTTP 响应保持 200，调用方检查事件内的业务码。 +- LangGraph 文本模型调用使用 `langchain-openai.ChatOpenAI` 的 ainvoke/astream； + 命名 profile 统一映射到 Chat Completions，温度未配置时不发送该参数， + EXTRA_BODY 保留原样。callbacks、metadata、tags 从图节点继承，每次尝试 + 标记运行名及 llm_attempt。流式 usage 默认不额外请求，兼容既有网关。 + 提前关闭流时显式关闭 provider 迭代器，避免等待 GC 后才释放 HTTP 连接。 - SDK 自动重试关闭，由应用层统一控制；默认首次尝试加 3 次重试， 每次尝试总超时 60 秒，退避 1、2、4 秒。单次 LLM 调用理论预算 4×60+1+2+4=247 秒（另有响应关闭的清理时间）。FastGPT SDK 的 diff --git a/docs/langgraph-backend-migration-plan.md b/docs/langgraph-backend-migration-plan.md index 48e181f..a7810b4 100644 --- a/docs/langgraph-backend-migration-plan.md +++ b/docs/langgraph-backend-migration-plan.md @@ -357,7 +357,7 @@ docs/langgraph/ | FastGPT 应用层配置（应用级系统提示词、思考模式开关）与线上部署差异 | 作业 4/5 验收对账前 | 移植代码、离线评测与差异记录 | | 评测案例与工作流导出的分歧（“对”分流语义、2004 走向）的权威口径 | 作业 5 验收前 | 白名单双走向、逐字比对测试、失败证据留档 | -## 当前进度与已知差异（数据库实现更新，模型评测为历史记录） +## 当前进度与已知差异（模型调用已迁移 ChatOpenAI，黄金评测为历史记录） ### 作业进度 @@ -373,9 +373,12 @@ docs/langgraph/ | 8. 真流式 | 已实现核心路径 | 模型增量、前缀校验、取消与提交回调；流式终止事件回归 | | 9—10 | 未完成 | 生产访问边界、容量验证与双后端等价性验收继续保留 | -本轮全量测试 315 通过，包含真实 SQLite/PostgreSQL 集成测试。提示词来源和 -逐字比对见 `test/agent/test_workflow_prompts.py`。下方真实模型结果是此前轮次， -本次数据库开发未重新运行 46 条真实模型黄金评测。 +数据库实现阶段全量测试 315 通过，包含真实 SQLite/PostgreSQL 集成测试。 +模型调用迁移为 `langchain-openai.ChatOpenAI` 后，全量回归 316 通过、13 跳过 +（本轮未配置隔离 PostgreSQL 测试库）。新增真实 HTTP 传输 mock 验证请求参数、 +流式关闭、取消、重试预算、图 callback 继承和共享模型实例的并发隔离。 +提示词来源和逐字比对见 `test/agent/test_workflow_prompts.py`。下方真实模型 +黄金评测结果是此前轮次，数据库开发和本次模型适配迁移均未重新运行 46 条评测。 ### 已落地的架构决策 @@ -415,7 +418,8 @@ src/agent/ collect.py 采集节点 + validated_reply（一次调用、一次修复预算） photo.py 拍照节点（四份提示词按 kind×mode 选择） confirm.py 核实节点（字段提取按 person_index 隔离） - llm.py LLMService Protocol / Fake / OpenAI(超时重试温度) / workflow fake + llm.py LLMService Protocol / Fake / ChatOpenAI(超时重试温度) / workflow fake + chat_model.py ChatOpenAI 流式迭代器的显式关闭与并发上下文隔离 fields.py 字段三态校验；clock.py 可控时钟 prompts/ 七份版本化提示词（来源 nodeId + 逐字原文） ``` diff --git a/requirements.txt b/requirements.txt index fd24525..c8f046e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -22,6 +22,8 @@ sqlalchemy pymysql PyYAML>=6.0 langgraph==1.2.9 +langchain-core==1.5.1 +langchain-openai==1.1.9 langgraph-checkpoint-sqlite==3.1.2 langgraph-checkpoint-postgres==3.2.0 psycopg[binary,pool]>=3.2,<4 diff --git a/src/agent/chat_model.py b/src/agent/chat_model.py new file mode 100644 index 0000000..26ac74e --- /dev/null +++ b/src/agent/chat_model.py @@ -0,0 +1,51 @@ +"""ChatOpenAI 流的显式生命周期管理。 + +当前 langchain-core 的 astream 不显式关闭嵌套的 provider 迭代器。 +提前停止消费（非法前缀、客户端断开）时，等待 GC 会延后 HTTP 连接释放。 +这里只跟踪并关闭原生迭代器，模型请求与 callbacks 仍由 ChatOpenAI 实现。 +""" + +from collections.abc import AsyncIterator +from contextvars import ContextVar + +from langchain_openai import ChatOpenAI +from langchain_openai.chat_models.base import BaseChatOpenAI + +_stream_iterators: ContextVar[list[AsyncIterator] | None] = ContextVar( + "chat_model_stream_iterators", default=None, +) + + +class ManagedChatOpenAI(ChatOpenAI): + async def astream(self, *args, **kwargs): + providers: list[AsyncIterator] = [] + source = super().astream(*args, **kwargs) + try: + while True: + # 每次推进时绑定，yield 前复原；兼容在不同任务中 anext/aclose。 + token = _stream_iterators.set(providers) + try: + chunk = await anext(source) + except StopAsyncIteration: + return + finally: + _stream_iterators.reset(token) + yield chunk + finally: + try: + await source.aclose() + finally: + for provider in reversed(providers): + await provider.aclose() + + def _astream(self, *args, **kwargs): + # ChatOpenAI 的协议路由也会创建一层未显式关闭的生成器。 + # 保留真正持有 HTTP 响应的原生 provider 迭代器。 + if self._use_responses_api({**kwargs, **self.model_kwargs}): + source = BaseChatOpenAI._astream_responses(self, *args, **kwargs) + else: + source = BaseChatOpenAI._astream(self, *args, **kwargs) + providers = _stream_iterators.get() + if providers is not None: + providers.append(source) + return source diff --git a/src/agent/llm.py b/src/agent/llm.py index 8b2c2a7..22daf80 100644 --- a/src/agent/llm.py +++ b/src/agent/llm.py @@ -2,8 +2,8 @@ fake 与真实 provider 实现同一 ``LLMService`` Protocol，节点不感知差异： - ``FakeLLMService``：脚本化固定输出，离线可重复，日常测试用它； -- ``OpenAILLMService``：真实 provider（TODO 关键代码）—— - 超时、取消、有限重试、模型配置。 +- ``OpenAILLMService``：通过 LangChain ChatOpenAI 调用 OpenAI 兼容接口， + 统一管理超时、取消、有限重试与模型配置。 ``build_llm_service`` 按命名提供商列表（Config.LLM_PROVIDERS / LLM_PROVIDER） 选择实现，见 core.config。 @@ -87,8 +87,9 @@ class FakeLLMService: class OpenAILLMService: - """单层重试：首次 + max_retries；每次尝试超时覆盖整个响应读取。 + """ChatOpenAI 适配；保留名称以兼容既有调用方。 + 单层重试：首次 + max_retries；每次尝试超时覆盖整个响应读取。 SDK 自动重试关闭。流式仅在首个文本 token 前重试，输出后失败终止。 默认 60 秒、3 次重试、1/2/4 秒退避：单次 LLM 调用预算上界 247 秒 （不含响应关闭的清理时间），不代表多次 LLM 调用的整轮 HTTP 超时。 @@ -115,40 +116,51 @@ class OpenAILLMService: def _get_client(self): if self._client is None: import httpx - from openai import AsyncOpenAI + from .chat_model import ManagedChatOpenAI from ..core.config import Config - options = { - "api_key": self._api_key, "timeout": self._timeout_seconds, - "max_retries": 0, - "http_client": httpx.AsyncClient(trust_env=Config.LLM_HTTP_TRUST_ENV), - } - if self._base_url: - options["base_url"] = self._base_url - self._client = AsyncOpenAI(**options) + self._client = ManagedChatOpenAI( + model=self._model, + api_key=self._api_key, + base_url=self._base_url or None, + timeout=self._timeout_seconds, + max_retries=0, + temperature=self._temperature, + extra_body=dict(self._extra_body) or None, + # 所有命名 profile 使用 Chat Completions，避免按模型名自动切协议。 + use_responses_api=False, + # 兼容网关未必支持 include_usage；不额外改变现有请求参数。 + stream_usage=False, + # 代理统一由下方 httpx 客户端和 LLM_HTTP_TRUST_ENV 控制。 + openai_proxy="", + http_client=httpx.Client(trust_env=Config.LLM_HTTP_TRUST_ENV), + http_async_client=httpx.AsyncClient(trust_env=Config.LLM_HTTP_TRUST_ENV), + ) return self._client async def aclose(self): if self._client is not None: - await self._client.close() - self._client = None + client = self._client + try: + await client.root_async_client.close() + finally: + try: + client.root_client.close() + finally: + self._client = None - def _request_args(self, system_prompt, user_message, history, *, stream=False): - options = { - "model": self._model, - "messages": [ - {"role": "system", "content": system_prompt}, - *[dict(message) for message in (history or [])], - {"role": "user", "content": user_message}, - ], - } - if stream: - options["stream"] = True - if self._temperature is not None: - options["temperature"] = self._temperature - if self._extra_body: - options["extra_body"] = dict(self._extra_body) - return options + @staticmethod + def _messages(system_prompt, user_message, history): + return [ + {"role": "system", "content": system_prompt}, + *[dict(message) for message in (history or [])], + {"role": "user", "content": user_message}, + ] + + @staticmethod + def _text(message): + # text 会只提取文本块，忽略 reasoning/tool/usage 等非正文内容。 + return str(message.text) @staticmethod def _retryable(exc): @@ -182,12 +194,15 @@ class OpenAILLMService: import asyncio client = self._get_client() - options = self._request_args(system_prompt, user_message, history) + messages = self._messages(system_prompt, user_message, history) for attempt in range(self._max_retries + 1): try: async with asyncio.timeout(self._timeout_seconds): - response = await client.chat.completions.create(**options) - content = response.choices[0].message.content if response.choices else None + response = await client.ainvoke( + messages, + config={"run_name": "llm.complete", "metadata": {"llm_attempt": attempt + 1}}, + ) + content = self._text(response) if not content: raise LLMAPIError("LLM 返回了空内容") return content @@ -205,18 +220,18 @@ class OpenAILLMService: from contextlib import suppress client = self._get_client() - options = self._request_args(system_prompt, user_message, history, stream=True) + messages = self._messages(system_prompt, user_message, history) emitted = False for attempt in range(self._max_retries + 1): response = None try: async with asyncio.timeout(self._timeout_seconds): - response = await client.chat.completions.create(**options) + response = client.astream( + messages, + config={"run_name": "llm.stream", "metadata": {"llm_attempt": attempt + 1}}, + ) async for event in response: - choices = getattr(event, "choices", None) - if not choices: - continue - token = getattr(getattr(choices[0], "delta", None), "content", None) + token = self._text(event) if token: emitted = True yield token @@ -229,10 +244,12 @@ class OpenAILLMService: if emitted or attempt == self._max_retries or not self._retryable(exc): raise self._error(exc) from exc finally: - close = getattr(response, "close", None) + close = getattr(response, "aclose", None) if close is not None: with suppress(Exception): - await asyncio.wait_for(close(), timeout=5) + # 在原任务中关闭，保留 callback/contextvars 上下文。 + async with asyncio.timeout(5): + await close() await self._retry_delay(attempt) diff --git a/test/agent/test_chat_openai_adapter.py b/test/agent/test_chat_openai_adapter.py new file mode 100644 index 0000000..e50bac2 --- /dev/null +++ b/test/agent/test_chat_openai_adapter.py @@ -0,0 +1,291 @@ +"""通过真实 ChatOpenAI/OpenAI SDK + HTTP mock 验证迁移边界。""" + +import asyncio +import json +from contextlib import aclosing +from typing import TypedDict +from unittest.mock import AsyncMock + +import httpx +import pytest +from langchain_core.callbacks import AsyncCallbackHandler +from langchain_core.messages import AIMessage +from langgraph.graph import END, START, StateGraph + +from src.agent.llm import LLMAPIError, LLMTimeoutError, OpenAILLMService +from src.core.config import Config + + +def _chunk(text): + data = { + 'id': 'response', 'object': 'chat.completion.chunk', 'created': 0, 'model': 'probe', + 'choices': [{'index': 0, 'delta': {'content': text}, 'finish_reason': None}], + } + return ('data: ' + json.dumps(data) + '\n\n').encode() + + +def _response(text, *, stream=False): + if stream: + return httpx.Response(200, headers={'content-type': 'text/event-stream'}, + content=_chunk(text) + b'data: [DONE]\n\n') + return httpx.Response(200, json={ + 'id': 'response', 'object': 'chat.completion', 'created': 0, 'model': 'probe', + 'choices': [{'index': 0, 'message': {'role': 'assistant', 'content': text}, + 'finish_reason': 'stop'}], + 'usage': {'prompt_tokens': 7, 'completion_tokens': 2, 'total_tokens': 9}, + }) + + +def _service(monkeypatch, handler, **kwargs): + monkeypatch.setattr(Config, 'LLM_HTTP_TRUST_ENV', False) + service = OpenAILLMService( + model='probe', api_key='offline-test', base_url='http://probe.local/v1', + extra_body={'reasoning_effort': 'none'}, **kwargs, + ) + # 只替换 HTTP 传输，模型工厂、消息转换、callbacks 和 SDK 全部真实执行。 + model = service._get_client() + model.http_async_client._transport = httpx.MockTransport(handler) + return service + + +@pytest.mark.asyncio +@pytest.mark.parametrize('stream', [False, True]) +@pytest.mark.parametrize('temperature', [None, 0.0]) +async def test_wire_contract_preserves_profile_messages_and_chat_completions(monkeypatch, stream, temperature): + monkeypatch.setenv('OPENAI_PROXY', 'http://invalid-proxy.local:1') + monkeypatch.setenv('HTTP_PROXY', 'http://invalid-proxy.local:1') + monkeypatch.setenv('HTTPS_PROXY', 'http://invalid-proxy.local:1') + requests = [] + def handler(request): + requests.append(request) + return _response('收到', stream=stream) + llm = _service(monkeypatch, handler, temperature=temperature) + http_async = llm._client.http_async_client + http_sync = llm._client.http_client + history = [{'role': 'user', 'content': '之前的输入'}, {'role': 'assistant', 'content': '之前的回复'}] + try: + if stream: + result = ''.join([t async for t in llm.stream_tokens('系统', '本轮', history=history)]) + else: + result = await llm.complete('系统', '本轮', history=history) + assert result == '收到' + assert len(requests) == 1 + request = requests[0] + payload = json.loads(request.content) + assert request.url.path == '/v1/chat/completions' + assert payload['model'] == 'probe' + assert payload['reasoning_effort'] == 'none' + assert payload['stream'] is stream + assert 'stream_options' not in payload + if temperature is None: + assert 'temperature' not in payload + else: + assert payload['temperature'] == temperature + assert payload['messages'] == [ + {'role': 'system', 'content': '系统'}, *history, {'role': 'user', 'content': '本轮'}, + ] + assert history == [{'role': 'user', 'content': '之前的输入'}, {'role': 'assistant', 'content': '之前的回复'}] + finally: + await llm.aclose() + assert http_async.is_closed and http_sync.is_closed + assert llm._client is None + await llm.aclose() + + +class TrackingBody(httpx.AsyncByteStream): + def __init__(self, text='partial', *, fail=False, emit=True): + self.text = text + self.fail = fail + self.emit = emit + self.closed = False + self.started = asyncio.Event() + self.release = asyncio.Event() + + async def __aiter__(self): + self.started.set() + if self.emit: + yield _chunk(self.text) + if self.fail: + raise httpx.ReadError('broken transport') + await self.release.wait() + yield _chunk('tail') + yield b'data: [DONE]\n\n' + + async def aclose(self): + self.closed = True + + +def _stream_response(body): + return httpx.Response(200, headers={'content-type': 'text/event-stream'}, stream=body) + + +@pytest.mark.asyncio +async def test_early_close_is_immediate_and_does_not_close_another_session(monkeypatch): + bodies = {} + def handler(request): + text = json.loads(request.content)['messages'][-1]['content'] + body = bodies[text] = TrackingBody(text) + return _stream_response(body) + llm = _service(monkeypatch, handler) + first = llm.stream_tokens('system', 'first') + second = llm.stream_tokens('system', 'second') + try: + # anext 运行在其他任务中，aclose 在当前任务中；上下文也必须安全复原。 + assert await asyncio.wait_for(anext(first), 5) == 'first' + assert await asyncio.wait_for(anext(second), 5) == 'second' + await first.aclose() + assert bodies['first'].closed + assert not bodies['second'].closed + bodies['second'].release.set() + assert [t async for t in second] == ['tail'] + assert bodies['second'].closed + finally: + await first.aclose() + await second.aclose() + await llm.aclose() + + +@pytest.mark.asyncio +async def test_transport_failure_after_text_closes_immediately_without_retry(monkeypatch): + body = TrackingBody(fail=True) + calls = [] + def handler(request): + calls.append(request) + return _stream_response(body) + llm = _service(monkeypatch, handler) + llm._retry_delay = AsyncMock() + tokens = [] + try: + with pytest.raises(LLMAPIError): + async for token in llm.stream_tokens('system', 'user'): + tokens.append(token) + assert tokens == ['partial'] + assert body.closed + assert len(calls) == 1 + llm._retry_delay.assert_not_awaited() + finally: + await llm.aclose() + + +@pytest.mark.asyncio +async def test_stream_cancellation_closes_http_response_before_returning(monkeypatch): + body = TrackingBody(emit=False) + llm = _service(monkeypatch, lambda request: _stream_response(body)) + llm._retry_delay = AsyncMock() + async def consume(): + async with aclosing(llm.stream_tokens('system', 'user')) as stream: + async for _ in stream: + pass + task = asyncio.create_task(consume()) + try: + await asyncio.wait_for(body.started.wait(), 5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert body.closed + llm._retry_delay.assert_not_awaited() + finally: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + await llm.aclose() + + +@pytest.mark.asyncio +async def test_stream_timeout_closes_each_attempt_before_retry(monkeypatch): + bodies = [] + def handler(request): + assert all(body.closed for body in bodies) + body = TrackingBody(emit=False) + bodies.append(body) + return _stream_response(body) + llm = _service(monkeypatch, handler, timeout_seconds=0.1, max_retries=2) + llm._retry_delay = AsyncMock() + try: + with pytest.raises(LLMTimeoutError): + async for _ in llm.stream_tokens('system', 'user'): + pass + assert len(bodies) == 3 + assert all(body.closed for body in bodies) + assert llm._retry_delay.await_count == 2 + finally: + await llm.aclose() + + +class TraceRecorder(AsyncCallbackHandler): + def __init__(self): + self.chains = {} + self.models = [] + self.results = [] + + async def on_chain_start(self, serialized, inputs, *, run_id, **kwargs): + self.chains[run_id] = kwargs.get('name') + + async def on_chat_model_start(self, serialized, messages, *, run_id, **kwargs): + self.models.append({'run_id': run_id, **kwargs}) + + async def on_llm_end(self, response, **kwargs): + self.results.append(response) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('stream', [False, True]) +async def test_model_callbacks_inherit_graph_parent_metadata_and_tags(monkeypatch, stream): + llm = _service(monkeypatch, lambda request: _response('ok', stream=stream)) + recorder = TraceRecorder() + class State(TypedDict): + text: str + async def dialogue(state): + if stream: + text = ''.join([t async for t in llm.stream_tokens('system', state['text'])]) + else: + text = await llm.complete('system', state['text']) + return {'text': text} + graph = StateGraph(State) + graph.add_node('dialogue', dialogue) + graph.add_edge(START, 'dialogue') + graph.add_edge('dialogue', END) + try: + result = await graph.compile().ainvoke({'text': 'hello'}, config={ + 'callbacks': [recorder], 'metadata': {'session_id': 'trace-session'}, 'tags': ['probe'], + }) + assert result['text'] == 'ok' + assert len(recorder.models) == 1 + model = recorder.models[0] + assert recorder.chains[model['parent_run_id']] == 'dialogue' + assert model['name'] == ('llm.stream' if stream else 'llm.complete') + assert model['metadata']['session_id'] == 'trace-session' + assert model['metadata']['llm_attempt'] == 1 + assert 'probe' in model['tags'] + assert len(recorder.results) == 1 + if not stream: + usage = recorder.results[0].generations[0][0].message.usage_metadata + assert usage['input_tokens'] == 7 and usage['output_tokens'] == 2 + finally: + await llm.aclose() + + +@pytest.mark.parametrize('content, expected', [ + ('正文', '正文'), + ([{'type': 'reasoning', 'reasoning': '内部内容'}, {'type': 'text', 'text': '正文'}], '正文'), + ([{'type': 'reasoning', 'reasoning': '内部内容'}], ''), +]) +def test_only_text_content_is_forwarded(content, expected): + assert OpenAILLMService._text(AIMessage(content=content)) == expected + + +@pytest.mark.asyncio +async def test_shared_model_serves_concurrent_calls_without_mixing_messages(monkeypatch): + async def handler(request): + payload = json.loads(request.content) + await asyncio.sleep(0) + return _response(payload['messages'][-1]['content'], stream=payload['stream']) + llm = _service(monkeypatch, handler) + async def request(index): + text = f'input-{index}' + if index % 2: + return ''.join([t async for t in llm.stream_tokens('system', text)]) + return await llm.complete('system', text) + try: + assert await asyncio.gather(*(request(i) for i in range(12))) == [f'input-{i}' for i in range(12)] + finally: + await llm.aclose() diff --git a/test/agent/test_llm_error_mapping.py b/test/agent/test_llm_error_mapping.py index 9b820c8..6226bef 100644 --- a/test/agent/test_llm_error_mapping.py +++ b/test/agent/test_llm_error_mapping.py @@ -37,9 +37,7 @@ def _service_with_failing_llm(exc: Exception, max_retries: int = 0): """complete 固定抛 exc 的 OpenAILLMService（client 打桩）。""" llm = OpenAILLMService(model="test", api_key="test", max_retries=max_retries) create = AsyncMock(side_effect=exc) - llm._client = SimpleNamespace( - chat=SimpleNamespace(completions=SimpleNamespace(create=create)) - ) + llm._client = SimpleNamespace(ainvoke=create) return llm, create diff --git a/test/agent/test_llm_stream_tokens.py b/test/agent/test_llm_stream_tokens.py index a94c3fc..57131d6 100644 --- a/test/agent/test_llm_stream_tokens.py +++ b/test/agent/test_llm_stream_tokens.py @@ -6,7 +6,9 @@ mock 客户端返回 token 事件流；验证消息构造与 complete 一致、 import asyncio from types import SimpleNamespace -from unittest.mock import AsyncMock +from unittest.mock import Mock + +from langchain_core.messages import AIMessageChunk import httpx import pytest @@ -22,9 +24,7 @@ from src.agent.llm import ( def _event(content): - return SimpleNamespace( - choices=[SimpleNamespace(delta=SimpleNamespace(content=content))] - ) + return AIMessageChunk(content=content) def _service_with_stream(events): @@ -44,10 +44,8 @@ def _service_with_stream(events): for item in self._items: yield item - create = AsyncMock(return_value=_Stream(events)) - llm._client = SimpleNamespace( - chat=SimpleNamespace(completions=SimpleNamespace(create=create)) - ) + create = Mock(return_value=_Stream(events)) + llm._client = SimpleNamespace(astream=create) return llm, create @@ -60,13 +58,11 @@ async def test_stream_tokens_yields_incrementally_and_forwards_kwargs(): ] assert tokens == ["你", "好"] # 空 delta 被跳过 - kwargs = create.call_args.kwargs - assert kwargs["model"] == "test-model" - assert kwargs["stream"] is True - assert kwargs["temperature"] == 0 - assert kwargs["extra_body"]["reasoning_effort"] == "none" # extra_body 透传 - assert kwargs["messages"][0] == {"role": "system", "content": "系统"} - assert kwargs["messages"][-1] == {"role": "user", "content": "输入"} + messages = create.call_args.args[0] + assert messages[0] == {"role": "system", "content": "系统"} + assert messages[1] == {"role": "assistant", "content": "h"} + assert messages[-1] == {"role": "user", "content": "输入"} + assert create.call_args.kwargs["config"]["run_name"] == "llm.stream" @pytest.mark.asyncio @@ -107,10 +103,8 @@ async def test_stream_tokens_times_out_on_stale_stream(): yield _event("好") llm = OpenAILLMService(model="m", api_key="k", timeout_seconds=0.2) - create = AsyncMock(return_value=_HangingStream()) - llm._client = SimpleNamespace( - chat=SimpleNamespace(completions=SimpleNamespace(create=create)) - ) + create = Mock(return_value=_HangingStream()) + llm._client = SimpleNamespace(astream=create) tokens = [] with pytest.raises(LLMTimeoutError): async for token in llm.stream_tokens("系统", "输入"): diff --git a/test/agent/test_workflow_prompts.py b/test/agent/test_workflow_prompts.py index a4865af..128f736 100644 --- a/test/agent/test_workflow_prompts.py +++ b/test/agent/test_workflow_prompts.py @@ -104,14 +104,14 @@ async def test_openai_adapter_preserves_message_roles(): from unittest.mock import AsyncMock from src.agent.llm import OpenAILLMService - create = AsyncMock(return_value=SimpleNamespace(choices=[ - SimpleNamespace(message=SimpleNamespace(content="1002收到。")) - ])) + from langchain_core.messages import AIMessage + + create = AsyncMock(return_value=AIMessage(content="1002收到。")) llm = OpenAILLMService(model="test", api_key="test", max_retries=0) - llm._client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=create))) + llm._client = SimpleNamespace(ainvoke=create) history = [{"role": "assistant", "content": "之前的问题"}] await llm.complete("系统提示词", "本轮输入", history=history) - assert create.call_args.kwargs["messages"] == [ + assert create.call_args.args[0] == [ {"role": "system", "content": "系统提示词"}, {"role": "assistant", "content": "之前的问题"}, {"role": "user", "content": "本轮输入"}, diff --git a/test/services/test_field_commit_and_retry.py b/test/services/test_field_commit_and_retry.py index 163c0a9..168f583 100644 --- a/test/services/test_field_commit_and_retry.py +++ b/test/services/test_field_commit_and_retry.py @@ -2,11 +2,13 @@ import asyncio import json from contextlib import aclosing from types import SimpleNamespace -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, Mock import httpx import pytest -from openai import AsyncOpenAI +from langchain_core.messages import AIMessageChunk + +from src.agent.chat_model import ManagedChatOpenAI from src.agent.llm import LLMAPIError, LLMRateLimitError, OpenAILLMService from src.agent.prompts import SINGLE_CONFIRM, render_prompt @@ -72,7 +74,10 @@ async def test_commit_is_not_lost_when_consumer_closes_after_graph_completes(): async def test_factory_disables_sdk_retries(): llm = OpenAILLMService(model='m', api_key='audit') try: - assert llm._get_client().max_retries == 0 + model = llm._get_client() + assert model.max_retries == 0 + assert model.root_async_client.max_retries == 0 + assert model.root_client.max_retries == 0 finally: await llm.aclose() @@ -101,8 +106,13 @@ async def test_actual_http_attempt_count_is_not_multiplied_by_sdk(stream, status return completion(stream) llm = OpenAILLMService(model='m', api_key='audit', max_retries=3, extra_body={'reasoning_effort': 'none'}) - llm._client = AsyncOpenAI(api_key='audit', base_url='http://audit/v1', max_retries=0, - http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler))) + llm._client = ManagedChatOpenAI( + model='m', api_key='audit', base_url='http://audit/v1', max_retries=0, + use_responses_api=False, stream_usage=False, temperature=None, + extra_body={'reasoning_effort': 'none'}, + http_client=httpx.Client(trust_env=False), + http_async_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) llm._retry_delay = AsyncMock() try: result = ''.join([t async for t in llm.stream_tokens('system', 'user')]) if stream else await llm.complete('system', 'user') @@ -120,19 +130,19 @@ async def test_stream_failure_after_first_token_closes_response_without_retry(): def __init__(self): self.closed = False def __aiter__(self): return self.generate() async def generate(self): - yield SimpleNamespace(choices=[SimpleNamespace(delta=SimpleNamespace(content='partial'))]) + yield AIMessageChunk(content='partial') raise OSError('broken socket') - async def close(self): self.closed = True + async def aclose(self): self.closed = True response = Response() - create = AsyncMock(return_value=response) + create = Mock(return_value=response) llm = OpenAILLMService(model='m', api_key='audit') - llm._client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=create))) + llm._client = SimpleNamespace(astream=create) tokens = [] with pytest.raises(LLMAPIError): async for token in llm.stream_tokens('system', 'user'): tokens.append(token) assert tokens == ['partial'] - assert create.await_count == 1 + assert create.call_count == 1 assert response.closed @@ -145,8 +155,12 @@ async def test_retry_budget_exhaustion_is_exact(stream): return httpx.Response(429, json={'error': {'message': 'retry'}}) llm = OpenAILLMService(model='m', api_key='audit', max_retries=2) llm._retry_delay = AsyncMock() - llm._client = AsyncOpenAI(api_key='audit', base_url='http://audit/v1', max_retries=0, - http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler))) + llm._client = ManagedChatOpenAI( + model='m', api_key='audit', base_url='http://audit/v1', max_retries=0, + use_responses_api=False, stream_usage=False, temperature=None, + http_client=httpx.Client(trust_env=False), + http_async_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) try: with pytest.raises(LLMRateLimitError): if stream: diff --git a/test/services/test_stream_retry_cleanup.py b/test/services/test_stream_retry_cleanup.py index 1f4cd22..2de87a6 100644 --- a/test/services/test_stream_retry_cleanup.py +++ b/test/services/test_stream_retry_cleanup.py @@ -1,7 +1,7 @@ import asyncio from contextlib import aclosing from types import SimpleNamespace -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, Mock import pytest @@ -69,16 +69,16 @@ async def test_timeout_covers_whole_attempt_and_retries_are_bounded(stream): async def generate(self): await asyncio.Future() yield None - async def close(self): self.closed = True - async def create(**kwargs): - if stream: - response = HangingResponse() - responses.append(response) - return response + async def aclose(self): self.closed = True + async def invoke(*args, **kwargs): await asyncio.Future() + def create_stream(*args, **kwargs): + response = HangingResponse() + responses.append(response) + return response llm = OpenAILLMService(model='m', api_key='audit', timeout_seconds=0.01, max_retries=2) - call = AsyncMock(side_effect=create) - llm._client = SimpleNamespace(chat=SimpleNamespace(completions=SimpleNamespace(create=call))) + call = Mock(side_effect=create_stream) if stream else AsyncMock(side_effect=invoke) + llm._client = SimpleNamespace(astream=call) if stream else SimpleNamespace(ainvoke=call) llm._retry_delay = AsyncMock() # 先加载 SDK 错误类；外层看门狗不测模块惰性导入耗时。 assert llm._retryable(TimeoutError()) @@ -89,6 +89,6 @@ async def test_timeout_covers_whole_attempt_and_retries_are_bounded(stream): async for _ in llm.stream_tokens('system', 'user'): pass else: await llm.complete('system', 'user') - assert call.await_count == 3 + assert call.call_count == 3 assert llm._retry_delay.await_count == 2 assert all(response.closed for response in responses)