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)