ZNJJ-api-server/examples/langgraph_smoke.py

"""LangGraph 脚手架冒烟脚本。

同一 thread 输入两次、另一个 thread 输入一次，打印轮数和文本。
预期轮数依次为 1、2、1；首轮为 welcome（1001 固定问候，不调模型），
第二轮起由注入的 workflow fake LLM 按提示词语义回复。

本脚本不导入 FastGPT、不读取其密钥；离线即可运行：

    python -m examples.langgraph_smoke
"""

import asyncio

from src.agent.graph import GraphDeps, build_graph, workflow_fake_llm


async def main() -> None:
    graph = build_graph()
    deps = GraphDeps(llm=workflow_fake_llm())

    async def ask(thread_id: str, text: str) -> None:
        # 只传本轮变化的字段；会话字段（turn_count/stage_code 等）
        # 由 checkpoint 保管。注意：对已有 checkpoint 的 thread 传
        # 完整初始状态会覆盖已累计的值。
        result = await graph.ainvoke(
            {"session_id": thread_id, "input_text": text},
            config={"configurable": {"thread_id": thread_id}},
            context=deps,
        )
        print(
            f"thread={thread_id} turn_count={result['turn_count']} "
            f"stage_code={result['stage_code']} text={result['response_text']}"
        )

    await ask("thread-a", "你好")
    await ask("thread-a", "发生了交通事故")
    await ask("thread-b", "你好")


if __name__ == "__main__":
    asyncio.run(main())