feat: add langchain dual-agent workflow

backend/src/kbqa/agents/__init__.py

@@ -0,0 +1 @@
+"""LangChain 1.x dual-agent workflow."""

backend/src/kbqa/agents/context.py

@@ -0,0 +1,17 @@
+from dataclasses import dataclass
+
+from kbqa.rag.retriever import KnowledgeRetriever
+
+
+@dataclass(frozen=True, slots=True)
+class ResearchContext:
+    retriever: KnowledgeRetriever
+    request_id: str
+
+
+@dataclass(frozen=True, slots=True)
+class ChatRunContext:
+    retriever: KnowledgeRetriever
+    request_id: str
+    session_id: str
+    assistant_message_id: str

backend/src/kbqa/agents/factory.py

@@ -0,0 +1,51 @@
+from dataclasses import dataclass
+from typing import Any
+
+from langchain.agents import create_agent
+from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
+
+from kbqa.agents.context import ChatRunContext, ResearchContext
+from kbqa.agents.llm import create_chat_model
+from kbqa.agents.middleware import RequireResearchMiddleware
+from kbqa.agents.prompts import MAIN_PROMPT, RESEARCH_PROMPT
+from kbqa.agents.tools import create_research_tool, search_knowledge_base
+from kbqa.config import Settings
+
+
+@dataclass(frozen=True, slots=True)
+class AgentBundle:
+    main_agent: Any
+    research_agent: Any
+
+
+def create_agent_bundle(settings: Settings) -> AgentBundle:
+    model = create_chat_model(settings)
+    research_agent = create_agent(
+        model=model,
+        tools=[search_knowledge_base],
+        system_prompt=RESEARCH_PROMPT,
+        context_schema=ResearchContext,
+        middleware=[
+            ModelCallLimitMiddleware(run_limit=4, exit_behavior="error"),
+            ToolCallLimitMiddleware(
+                tool_name="search_knowledge_base",
+                run_limit=settings.max_research_searches,
+                exit_behavior="continue",
+            ),
+        ],
+        name="research-agent",
+    )
+    research_tool = create_research_tool(research_agent)
+    main_agent = create_agent(
+        model=model,
+        tools=[research_tool],
+        system_prompt=MAIN_PROMPT,
+        context_schema=ChatRunContext,
+        middleware=[
+            RequireResearchMiddleware(),
+            ModelCallLimitMiddleware(run_limit=3, exit_behavior="error"),
+            ToolCallLimitMiddleware(tool_name="research", run_limit=1, exit_behavior="error"),
+        ],
+        name="main-agent",
+    )
+    return AgentBundle(main_agent=main_agent, research_agent=research_agent)

backend/src/kbqa/agents/llm.py

@@ -0,0 +1,16 @@
+from langchain_openai import ChatOpenAI
+
+from kbqa.config import Settings, validate_live_settings
+
+
+def create_chat_model(settings: Settings) -> ChatOpenAI:
+    validate_live_settings(settings)
+    return ChatOpenAI(
+        api_key=settings.dashscope_api_key,
+        base_url=settings.dashscope_base_url,
+        model=settings.chat_model,
+        temperature=0.2,
+        streaming=True,
+        timeout=120.0,
+        max_retries=2,
+    )

backend/src/kbqa/agents/middleware.py

@@ -0,0 +1,23 @@
+from collections.abc import Awaitable, Callable
+
+from langchain.agents.middleware import AgentMiddleware, ModelRequest, ModelResponse
+from langchain.messages import HumanMessage, ToolMessage
+
+
+class RequireResearchMiddleware(AgentMiddleware):
+    @staticmethod
+    def _has_current_research(messages: list) -> bool:
+        for message in reversed(messages):
+            if isinstance(message, ToolMessage) and message.name == "research":
+                return True
+            if isinstance(message, HumanMessage):
+                return False
+        return False
+
+    async def awrap_model_call(
+        self,
+        request: ModelRequest,
+        handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
+    ) -> ModelResponse:
+        tool_choice = "none" if self._has_current_research(request.messages) else "required"
+        return await handler(request.override(tool_choice=tool_choice))

backend/src/kbqa/agents/prompts.py

@@ -0,0 +1,18 @@
+RESEARCH_PROMPT = """你是知识库研究员，只能使用 search_knowledge_base 工具返回的资料。
+
+规则：
+1. 根据问题设计检索词；首次证据不足时可改写检索词，最多检索三次。
+2. 不得使用模型自身知识补充事实，不得伪造文档或引用。
+3. 汇总检索资料中的一致结论、差异和缺口；你不直接面向最终用户回答。
+4. 没有可靠资料时明确写出“未检索到足够证据”。
+"""
+
+MAIN_PROMPT = """你是严格知识库问答助手。你必须先调用 research 工具，再回答用户。
+
+规则：
+1. 只能依据 research 返回的编号证据回答，不使用模型自身知识补充事实。
+2. 每个知识性结论后使用 [S1]、[S2] 形式标注真实来源。
+3. 只允许使用 research 明确提供的来源标签；不得编造标签。
+4. 回答清晰、直接，区分确定事实与资料中的限制。
+5. 如果资料不足，明确说明知识库中没有足够资料。
+"""

backend/src/kbqa/agents/tools.py

@@ -0,0 +1,116 @@
+import json
+from collections.abc import Sequence
+from typing import Any
+
+from langchain.messages import ToolMessage
+from langchain.tools import ToolRuntime, tool
+from langgraph.config import get_stream_writer
+
+from kbqa.agents.context import ChatRunContext, ResearchContext
+from kbqa.rag.types import SourceCandidate
+
+
+def serialize_source(source: SourceCandidate) -> dict[str, Any]:
+    return {
+        "chunk_id": source.chunk_id,
+        "document_id": source.document_id,
+        "filename": source.filename,
+        "order_index": source.order_index,
+        "page_number": source.page_number,
+        "content": source.content,
+        "score": source.score,
+    }
+
+
+@tool(response_format="content_and_artifact")
+async def search_knowledge_base(
+    query: str,
+    runtime: ToolRuntime[ResearchContext],
+) -> tuple[str, list[dict[str, Any]]]:
+    """搜索本地知识库并返回与问题相关的文档片段。
+
+    Args:
+        query: 简洁、独立、适合语义检索的查询语句。
+    """
+    sources = await runtime.context.retriever.retrieve(query, top_k=5)
+    if not sources:
+        return "未检索到相关知识库片段。", []
+    artifact = [serialize_source(source) for source in sources]
+    content = "\n\n".join(
+        f"检索片段 {index + 1}（{source.filename}，第 {source.order_index + 1} 段）：\n"
+        f"{source.content}"
+        for index, source in enumerate(sources)
+    )
+    return content, artifact
+
+
+def _message_text(content: Any) -> str:
+    if isinstance(content, str):
+        return content
+    if isinstance(content, Sequence):
+        texts: list[str] = []
+        for block in content:
+            if isinstance(block, str):
+                texts.append(block)
+            elif isinstance(block, dict) and block.get("type") == "text":
+                texts.append(str(block.get("text", "")))
+        return "".join(texts)
+    return str(content)
+
+
+def create_research_tool(research_agent):
+    @tool(response_format="content_and_artifact")
+    async def research(
+        question: str,
+        runtime: ToolRuntime[ChatRunContext],
+    ) -> tuple[str, list[dict[str, Any]]]:
+        """调用研究 Agent 检索知识库并整理可引用证据。
+
+        Args:
+            question: 结合会话语境改写后的独立知识问题。
+        """
+        result = await research_agent.ainvoke(
+            {"messages": [{"role": "user", "content": question}]},
+            context=ResearchContext(
+                retriever=runtime.context.retriever,
+                request_id=runtime.context.request_id,
+            ),
+            config={
+                "recursion_limit": 12,
+                "tags": ["research-agent"],
+                "metadata": {"request_id": runtime.context.request_id},
+            },
+        )
+        by_chunk: dict[str, dict[str, Any]] = {}
+        for message in result["messages"]:
+            if not isinstance(message, ToolMessage) or message.name != "search_knowledge_base":
+                continue
+            for source in message.artifact or []:
+                chunk_id = str(source["chunk_id"])
+                current = by_chunk.get(chunk_id)
+                if current is None or float(source["score"]) > float(current["score"]):
+                    by_chunk[chunk_id] = dict(source)
+        sources = sorted(by_chunk.values(), key=lambda item: float(item["score"]), reverse=True)[:8]
+        writer = get_stream_writer()
+        if not sources:
+            writer({"kind": "no_evidence"})
+            return "知识库中没有足够资料。", []
+        labeled = []
+        evidence_blocks: list[str] = []
+        for index, source in enumerate(sources, start=1):
+            labeled_source = {**source, "label": f"S{index}"}
+            labeled.append(labeled_source)
+            evidence_blocks.append(
+                f"[S{index}] {source['filename']} 第 {int(source['order_index']) + 1} 段\n"
+                f"{source['content']}"
+            )
+        writer({"kind": "sources", "sources": labeled})
+        summary = _message_text(result["messages"][-1].content)
+        tool_content = f"研究员摘要：\n{summary}\n\n可引用证据：\n" + "\n\n".join(evidence_blocks)
+        return tool_content, labeled
+
+    return research
+
+
+def artifact_json(artifact: list[dict[str, Any]]) -> str:
+    return json.dumps(artifact, ensure_ascii=False, separators=(",", ":"))