相关资源

• 查看示例代码：docs/langgraph-learning/examples/01_real_model_agent.py

  """Chapter 1: one real model call inside a minimal LangGraph agent."""
  
  import asyncio
  
  from dotenv import load_dotenv
  from langchain.chat_models import init_chat_model
  from langchain_core.messages import HumanMessage
  from langgraph.graph import END, START, MessagesState, StateGraph
  from langgraph.runtime import Runtime
  
  from open_deep_research.configuration import Configuration
  
  
  load_dotenv()
  
  
  configurable_model = init_chat_model(
      configurable_fields=("model", "max_tokens", "api_key"),
  )
  
  
  async def answer(state: MessagesState, runtime: Runtime[Configuration]):
      """Call the configured real model and append its reply to message state."""
      settings = runtime.context
      model = configurable_model.with_config(
          {
              "configurable": {
                  "model": settings.research_model,
                  "max_tokens": 80,
              },
              "tags": ["langsmith:nostream"],
          }
      )
      response = await model.ainvoke(state["messages"])
      return {"messages": [response]}
  
  
  async def main():
      graph = (
          StateGraph(MessagesState, context_schema=Configuration)
          .add_node("answer", answer)
          .add_edge(START, "answer")
          .add_edge("answer", END)
          .compile()
      )
      result = await graph.ainvoke(
          {
              "messages": [
                  HumanMessage(content="只用一句中文说明 LangGraph 的作用。"),
              ]
          },
          context=Configuration.from_env(),
      )
      print(result["messages"][-1].content)
  
  
  if __name__ == "__main__":
      asyncio.run(main())
  

• 查看示例代码：docs/langgraph-learning/examples/02_state_and_command.py

  """Chapter 2: state reducers and Command routing with one real model call."""
  
  import asyncio
  import operator
  from typing import Annotated, Literal
  
  from dotenv import load_dotenv
  from langchain.chat_models import init_chat_model
  from langchain_core.messages import HumanMessage
  from langgraph.graph import END, START, MessagesState, StateGraph
  from langgraph.runtime import Runtime
  from langgraph.types import Command
  
  from open_deep_research.configuration import Configuration
  
  
  load_dotenv()
  
  
  class LearningState(MessagesState):
      """Conversation state plus an append-only record of graph routing."""
  
      route_log: Annotated[list[str], operator.add]
  
  
  configurable_model = init_chat_model(
      configurable_fields=("model", "max_tokens", "api_key"),
  )
  
  
  async def answer(
      state: LearningState,
      runtime: Runtime[Configuration],
  ) -> Command[Literal["finish"]]:
      """Call the real model, update state, then choose the next node."""
      settings = runtime.context
      model = configurable_model.with_config(
          {
              "configurable": {
                  "model": settings.research_model,
                  "max_tokens": 80,
              },
              "tags": ["langsmith:nostream"],
          }
      )
      response = await model.ainvoke(state["messages"])
      return Command(
          update={
              "messages": [response],
              "route_log": ["answer -> finish"],
          },
          goto="finish",
      )
  
  
  def finish(_: LearningState):
      """Append one local state update, then use its static edge to end."""
      return {"route_log": ["finish -> END"]}
  
  
  async def main():
      graph = (
          StateGraph(LearningState, context_schema=Configuration)
          .add_node("answer", answer)
          .add_node("finish", finish)
          .add_edge(START, "answer")
          .add_edge("finish", END)
          .compile()
      )
      result = await graph.ainvoke(
          {
              "messages": [
                  HumanMessage(content="只用一句中文解释状态机为什么适合 Agent。"),
              ],
              "route_log": [],
          },
          context=Configuration.from_env(),
      )
      print(f"消息数: {len(result['messages'])}")
      print("路由: " + " | ".join(result["route_log"]))
      print("答复: " + str(result["messages"][-1].content))
  
  
  if __name__ == "__main__":
      asyncio.run(main())
  

• 查看示例代码：docs/langgraph-learning/examples/03_structured_output.py

  """Chapter 3: one real structured-output call inside a minimal graph."""
  
  import asyncio
  
  from dotenv import load_dotenv
  from langchain.chat_models import init_chat_model
  from langchain_core.messages import AIMessage, HumanMessage
  from langgraph.graph import END, START, MessagesState, StateGraph
  from langgraph.runtime import Runtime
  from pydantic import BaseModel, Field
  
  from open_deep_research.configuration import Configuration
  
  
  load_dotenv()
  
  
  class TopicBrief(BaseModel):
      """A tiny schema for learning structured output."""
  
      title: str = Field(description="A short Chinese title.")
      research_question: str = Field(description="One focused Chinese research question.")
      needs_tools: bool = Field(description="Whether external search/tools are needed.")
  
  
  configurable_model = init_chat_model(
      configurable_fields=("model", "max_tokens", "api_key"),
  )
  
  
  async def make_brief(state: MessagesState, runtime: Runtime[Configuration]):
      settings = runtime.context
      model = configurable_model.with_config(
          {
              "configurable": {
                  "model": settings.research_model,
                  "max_tokens": 160,
              },
              "tags": ["langsmith:nostream"],
          }
      )
      structured_model = model.with_structured_output(TopicBrief)
      brief = await structured_model.ainvoke(state["messages"])
      return {"messages": [AIMessage(content=brief.model_dump_json(ensure_ascii=False))]}
  
  
  async def main():
      graph = (
          StateGraph(MessagesState, context_schema=Configuration)
          .add_node("make_brief", make_brief)
          .add_edge(START, "make_brief")
          .add_edge("make_brief", END)
          .compile()
      )
      result = await graph.ainvoke(
          {
              "messages": [
                  HumanMessage(content="我想研究 LangGraph 的状态管理。"),
              ]
          },
          context=Configuration.from_env(),
      )
      print(result["messages"][-1].content)
  
  
  if __name__ == "__main__":
      asyncio.run(main())
  

• 查看示例代码：docs/langgraph-learning/examples/04_tool_loop.py

  """Chapter 4: a tiny real ReAct loop with one bound tool."""
  
  import asyncio
  
  from dotenv import load_dotenv
  from langchain.chat_models import init_chat_model
  from langchain_core.messages import HumanMessage, ToolMessage
  from langchain_core.tools import tool
  from langgraph.graph import START, MessagesState, StateGraph
  from langgraph.runtime import Runtime
  from langgraph.types import Command
  
  from open_deep_research.configuration import Configuration
  
  
  load_dotenv()
  
  
  @tool
  def multiply_by_two(value: int) -> str:
      """Multiply the input integer by two."""
      return str(value * 2)
  
  
  configurable_model = init_chat_model(
      configurable_fields=("model", "max_tokens", "api_key"),
  )
  
  
  async def agent(state: MessagesState, runtime: Runtime[Configuration]):
      settings = runtime.context
      model = configurable_model.with_config(
          {
              "configurable": {
                  "model": settings.research_model,
                  "max_tokens": 120,
              },
              "tags": ["langsmith:nostream"],
          }
      ).bind_tools([multiply_by_two])
      response = await model.ainvoke(state["messages"])
      if response.tool_calls:
          return Command(update={"messages": [response]}, goto="run_tools")
      return {"messages": [response]}
  
  
  async def run_tools(state: MessagesState):
      last_message = state["messages"][-1]
      outputs = []
      for tool_call in last_message.tool_calls:
          result = multiply_by_two.invoke(tool_call["args"])
          outputs.append(
              ToolMessage(
                  content=result,
                  name=tool_call["name"],
                  tool_call_id=tool_call["id"],
              )
          )
      return Command(update={"messages": outputs}, goto="agent")
  
  
  async def main():
      graph = (
          StateGraph(MessagesState, context_schema=Configuration)
          .add_node("agent", agent)
          .add_node("run_tools", run_tools)
          .add_edge(START, "agent")
          .add_edge("run_tools", "agent")
          .compile()
      )
      result = await graph.ainvoke(
          {
              "messages": [
                  HumanMessage(
                      content="请使用可用工具计算 21 的两倍，然后只用一句中文给出答案。"
                  )
              ]
          },
          context=Configuration.from_env(),
      )
      print(f"消息数: {len(result['messages'])}")
      print("最终答复: " + str(result["messages"][-1].content))
  
  
  if __name__ == "__main__":
      asyncio.run(main())
  

• 查看示例代码：docs/langgraph-learning/examples/05_search_and_mcp.py

  """Chapter 5: assemble project tools, then run one safe real tool loop."""
  
  import asyncio
  
  from dotenv import load_dotenv
  from langchain.chat_models import init_chat_model
  from langchain_core.messages import HumanMessage, ToolMessage
  
  from open_deep_research.configuration import Configuration
  from open_deep_research.utils import get_all_tools, get_api_key_for_model
  
  
  load_dotenv()
  
  
  configurable_model = init_chat_model(
      configurable_fields=("model", "max_tokens", "api_key"),
  )
  
  
  def tool_name(tool):
      return tool.name if hasattr(tool, "name") else tool.get("name", "web_search")
  
  
  async def main():
      settings = Configuration.from_env()
      tool_context = Configuration.model_validate(
          {**settings.model_dump(), "search_api": "none"}
      )
      tools = await get_all_tools(tool_context)
      print("可用工具: " + ", ".join(tool_name(tool) for tool in tools))
  
      think_tool = next(tool for tool in tools if tool_name(tool) == "think_tool")
      model = configurable_model.with_config(
          {
              "configurable": {
                  "model": settings.research_model,
                  "max_tokens": 160,
                  "api_key": get_api_key_for_model(settings.research_model, settings),
              },
              "tags": ["langsmith:nostream"],
          }
      ).bind_tools([think_tool])
  
      messages = [
          HumanMessage(
              content=(
                  "请先调用 think_tool，反思学习搜索与 MCP 时最该关注的一个边界，"
                  "然后用一句中文总结。"
              )
          )
      ]
      first_response = await model.ainvoke(messages)
      messages.append(first_response)
      print(f"工具调用数: {len(first_response.tool_calls)}")
  
      tool_outputs = []
      for tool_call in first_response.tool_calls:
          result = think_tool.invoke(tool_call["args"])
          tool_outputs.append(
              ToolMessage(
                  content=result,
                  name=tool_call["name"],
                  tool_call_id=tool_call["id"],
              )
          )
  
      final_response = await model.ainvoke(messages + tool_outputs)
      print("最终答复: " + str(final_response.content))
  
  
  if __name__ == "__main__":
      asyncio.run(main())
  

• 查看示例代码：docs/langgraph-learning/examples/06_subgraphs_and_concurrency.py

  """Chapter 6: invoke two compiled subgraphs concurrently with real models."""
  
  import asyncio
  from typing import TypedDict
  
  from dotenv import load_dotenv
  from langchain.chat_models import init_chat_model
  from langchain_core.messages import HumanMessage
  from langgraph.graph import END, START, StateGraph
  from langgraph.runtime import Runtime
  
  from open_deep_research.configuration import Configuration
  
  
  load_dotenv()
  
  
  class ResearcherState(TypedDict):
      topic: str
      summary: str
  
  
  class ResearcherOutput(TypedDict):
      summary: str
  
  
  class ParentState(TypedDict):
      topics: list[str]
      summaries: list[str]
  
  
  configurable_model = init_chat_model(
      configurable_fields=("model", "max_tokens", "api_key"),
  )
  
  
  async def summarize_topic(
      state: ResearcherState,
      runtime: Runtime[Configuration],
  ):
      settings = runtime.context
      model = configurable_model.with_config(
          {
              "configurable": {
                  "model": settings.research_model,
                  "max_tokens": 120,
              },
              "tags": ["langsmith:nostream"],
          }
      )
      response = await model.ainvoke(
          [
              HumanMessage(
                  content=f"只用一句中文概括这个 LangGraph 学习主题: {state['topic']}"
              )
          ]
      )
      return {"summary": str(response.content)}
  
  
  researcher_graph = (
      StateGraph(
          ResearcherState,
          context_schema=Configuration,
          input_schema=ResearcherState,
          output_schema=ResearcherOutput,
      )
      .add_node("summarize_topic", summarize_topic)
      .add_edge(START, "summarize_topic")
      .add_edge("summarize_topic", END)
      .compile()
  )
  
  
  async def run_researchers(
      state: ParentState,
      runtime: Runtime[Configuration],
  ):
      results = await asyncio.gather(
          *(
              researcher_graph.ainvoke(
                  {"topic": topic},
                  context=runtime.context,
              )
              for topic in state["topics"]
          )
      )
      return {"summaries": [result["summary"] for result in results]}
  
  
  async def main():
      parent_graph = (
          StateGraph(ParentState, context_schema=Configuration)
          .add_node("run_researchers", run_researchers)
          .add_edge(START, "run_researchers")
          .add_edge("run_researchers", END)
          .compile()
      )
      result = await parent_graph.ainvoke(
          {
              "topics": ["子图隔离状态", "asyncio.gather 并发调用"],
              "summaries": [],
          },
          context=Configuration.from_env(),
      )
      print(f"子图数: {len(result['summaries'])}")
      for summary in result["summaries"]:
          print("- " + summary)
  
  
  if __name__ == "__main__":
      asyncio.run(main())
  

• 查看示例代码：docs/langgraph-learning/examples/07_persistence_streaming_observability.py

  """Chapter 7: persist one thread, stream events, inspect saved state."""
  
  import asyncio
  
  from dotenv import load_dotenv
  from langchain.chat_models import init_chat_model
  from langchain_core.messages import HumanMessage
  from langgraph.checkpoint.memory import InMemorySaver
  from langgraph.graph import END, START, MessagesState, StateGraph
  from langgraph.runtime import Runtime
  
  from open_deep_research.configuration import Configuration
  
  
  load_dotenv()
  
  
  configurable_model = init_chat_model(
      configurable_fields=("model", "max_tokens", "api_key"),
  )
  
  
  async def reply(state: MessagesState, runtime: Runtime[Configuration]):
      settings = runtime.context
      model = configurable_model.with_config(
          {
              "configurable": {
                  "model": settings.research_model,
                  "max_tokens": 120,
              },
              "tags": ["langsmith:nostream"],
          }
      )
      response = await model.ainvoke(state["messages"])
      return {"messages": [response]}
  
  
  memory = InMemorySaver()
  graph = (
      StateGraph(MessagesState, context_schema=Configuration)
      .add_node("reply", reply)
      .add_edge(START, "reply")
      .add_edge("reply", END)
      .compile(checkpointer=memory)
  )
  
  
  async def count_events(input_value, config, context):
      count = 0
      stream = await graph.astream_events(
          input_value,
          config=config,
          context=context,
          version="v3",
      )
      async for _event in stream:
          count += 1
      return count
  
  
  async def main():
      config = {"configurable": {"thread_id": "learning-thread-1"}}
      context = Configuration.from_env()
  
      first_input = {
          "messages": [HumanMessage(content="你好，我叫老李。请记住我的名字。")]
      }
      event_count = await count_events(first_input, config, context)
  
      second_result = await graph.ainvoke(
          {
              "messages": [HumanMessage(content="我刚才说我叫什么？只用一句中文回答。")]
          },
          config=config,
          context=context,
      )
      snapshot = graph.get_state(config)
  
      print(f"事件数: {event_count}")
      print("第二轮答复: " + str(second_result["messages"][-1].content))
      print(f"持久化消息数: {len(snapshot.values['messages'])}")
  
  
  if __name__ == "__main__":
      asyncio.run(main())
  

• 查看示例代码：docs/langgraph-learning/examples/08_integrated_mini_researcher.py

  """Chapter 8: a compact mini deep researcher using the prior chapters."""
  
  import asyncio
  import operator
  from typing import Annotated, Literal
  
  from dotenv import load_dotenv
  from langchain.chat_models import init_chat_model
  from langchain_core.messages import HumanMessage, MessageLikeRepresentation, ToolMessage
  from langchain_core.tools import tool
  from langgraph.checkpoint.memory import InMemorySaver
  from langgraph.graph import END, START, MessagesState, StateGraph
  from langgraph.runtime import Runtime
  from langgraph.types import Command
  from pydantic import BaseModel, Field
  from typing_extensions import TypedDict
  
  from open_deep_research.configuration import Configuration
  
  
  load_dotenv()
  
  
  class MiniPlan(BaseModel):
      """A small structured research plan."""
  
      research_brief: str = Field(description="One focused Chinese research brief.")
      topics: list[str] = Field(
          description="Exactly two focused Chinese subtopics.",
          min_length=2,
          max_length=2,
      )
  
  
  class MiniState(MessagesState):
      research_brief: str
      topics: list[str]
      summaries: list[str]
      final_report: str
      route_log: Annotated[list[str], operator.add]
  
  
  class ResearcherState(TypedDict):
      topic: str
      researcher_messages: Annotated[list[MessageLikeRepresentation], operator.add]
      summary: str
  
  
  class ResearcherOutput(TypedDict):
      summary: str
  
  
  configurable_model = init_chat_model(
      configurable_fields=("model", "max_tokens", "api_key"),
  )
  
  
  def configured_model(settings: Configuration, max_tokens: int = 180):
      return configurable_model.with_config(
          {
              "configurable": {
                  "model": settings.research_model,
                  "max_tokens": max_tokens,
              },
              "tags": ["langsmith:nostream"],
          }
      )
  
  
  def text_content(content) -> str:
      if isinstance(content, list):
          return "".join(
              part.get("text", str(part)) if isinstance(part, dict) else str(part)
              for part in content
          )
      return str(content)
  
  
  @tool
  def record_learning_boundary(boundary: str) -> str:
      """Record one learning boundary before summarizing a research topic."""
      return f"已记录边界: {boundary}"
  
  
  async def researcher_agent(
      state: ResearcherState,
      runtime: Runtime[Configuration],
  ) -> Command[Literal["researcher_tools", "__end__"]]:
      model = configured_model(runtime.context)
      has_tool_result = any(
          getattr(message, "type", None) == "tool"
          for message in state.get("researcher_messages", [])
      )
      if not has_tool_result:
          response = await model.bind_tools(
              [record_learning_boundary],
              tool_choice="record_learning_boundary",
          ).ainvoke(
              [
                  HumanMessage(
                      content=(
                          f"研究主题: {state['topic']}。先调用工具记录一个学习边界。"
                      )
                  )
              ]
          )
          return Command(
              update={"researcher_messages": [response]},
              goto="researcher_tools",
          )
  
      response = await model.ainvoke(
          state["researcher_messages"]
          + [HumanMessage(content=f"基于工具记录，用一句中文总结: {state['topic']}")]
      )
      return Command(
          update={"researcher_messages": [response], "summary": text_content(response.content)},
          goto=END,
      )
  
  
  async def researcher_tools(
      state: ResearcherState,
  ) -> Command[Literal["researcher_agent"]]:
      last_message = state["researcher_messages"][-1]
      outputs = []
      for tool_call in last_message.tool_calls:
          outputs.append(
              ToolMessage(
                  content=record_learning_boundary.invoke(tool_call["args"]),
                  name=tool_call["name"],
                  tool_call_id=tool_call["id"],
              )
          )
      return Command(update={"researcher_messages": outputs}, goto="researcher_agent")
  
  
  researcher_graph = (
      StateGraph(
          ResearcherState,
          context_schema=Configuration,
          input_schema=ResearcherState,
          output_schema=ResearcherOutput,
      )
      .add_node("researcher_agent", researcher_agent)
      .add_node("researcher_tools", researcher_tools)
      .add_edge(START, "researcher_agent")
      .compile()
  )
  
  
  async def make_plan(
      state: MiniState,
      runtime: Runtime[Configuration],
  ) -> Command[Literal["run_researchers"]]:
      model = configured_model(runtime.context).with_structured_output(MiniPlan)
      plan = await model.ainvoke(
          state["messages"]
          + [
              HumanMessage(
                  content=(
                      "把用户学习目标整理成一个 research_brief 和两个子主题，"
                      "必须聚焦当前 open_deep_research 项目。"
                  )
              )
          ]
      )
      return Command(
          update={
              "research_brief": plan.research_brief,
              "topics": plan.topics,
              "route_log": ["make_plan -> run_researchers"],
          },
          goto="run_researchers",
      )
  
  
  async def run_researchers(
      state: MiniState,
      runtime: Runtime[Configuration],
  ) -> Command[Literal["write_final"]]:
      results = await asyncio.gather(
          *(
              researcher_graph.ainvoke(
                  {
                      "topic": topic,
                      "researcher_messages": [],
                      "summary": "",
                  },
                  context=runtime.context,
              )
              for topic in state["topics"]
          )
      )
      return Command(
          update={
              "summaries": [result["summary"] for result in results],
              "route_log": ["run_researchers -> write_final"],
          },
          goto="write_final",
      )
  
  
  async def write_final(state: MiniState, runtime: Runtime[Configuration]):
      response = await configured_model(runtime.context, max_tokens=240).ainvoke(
          [
              HumanMessage(
                  content=(
                      "请用三句中文写一个迷你研究报告。\n"
                      f"研究 brief: {state['research_brief']}\n"
                      f"研究摘要: {state['summaries']}"
                  )
              )
          ]
      )
      return {
          "final_report": text_content(response.content),
          "messages": [response],
          "route_log": ["write_final -> END"],
      }
  
  
  memory = InMemorySaver()
  mini_researcher = (
      StateGraph(MiniState, context_schema=Configuration)
      .add_node("make_plan", make_plan)
      .add_node("run_researchers", run_researchers)
      .add_node("write_final", write_final)
      .add_edge(START, "make_plan")
      .add_edge("write_final", END)
      .compile(checkpointer=memory)
  )
  
  
  async def main():
      config = {"configurable": {"thread_id": "mini-researcher-learning"}}
      context = Configuration.from_env()
      input_value = {
          "messages": [
              HumanMessage(
                  content=(
                      "我想通过 open_deep_research 学会 LangGraph 的工具循环和子图并发。"
                  )
              )
          ],
          "route_log": [],
          "summaries": [],
      }
  
      event_count = 0
      stream = await mini_researcher.astream_events(
          input_value,
          config=config,
          context=context,
          version="v3",
      )
      async for _event in stream:
          event_count += 1
  
      snapshot = mini_researcher.get_state(config)
      values = snapshot.values
      print(f"事件数: {event_count}")
      print("主题: " + " | ".join(values["topics"]))
      print("路由: " + " | ".join(values["route_log"]))
      print("摘要数: " + str(len(values["summaries"])))
      print("最终报告: " + values["final_report"])
      print("持久化消息数: " + str(len(values["messages"])))
  
  
  if __name__ == "__main__":
      asyncio.run(main())
  

• 查看示例代码：docs/langgraph-learning/examples/10_multi_agent_handoff.py

  """Chapter 15: structured handoff between a coordinator and researcher subgraphs."""
  
  import asyncio
  from typing import TypedDict
  
  from dotenv import load_dotenv
  from langchain.chat_models import init_chat_model
  from langchain_core.messages import HumanMessage
  from langgraph.graph import END, START, StateGraph
  from langgraph.runtime import Runtime
  from pydantic import BaseModel, Field
  
  from open_deep_research.configuration import Configuration
  
  
  load_dotenv()
  
  
  class Subtask(BaseModel):
      """A coordinator-to-researcher handoff contract."""
  
      topic: str = Field(description="One focused research topic in Chinese.")
      question: str = Field(description="The exact question this researcher must answer.")
      expected_evidence: str = Field(description="What evidence or explanation to return.")
  
  
  class ResearchPlan(BaseModel):
      """Coordinator output that decides the subgraph inputs."""
  
      brief: str = Field(description="A concise restatement of the user's goal.")
      tasks: list[Subtask] = Field(
          description="Exactly two independent research tasks.", min_length=2, max_length=2
      )
  
  
  class Finding(BaseModel):
      """A researcher-to-coordinator handoff contract."""
  
      topic: str
      answer: str = Field(description="A factual answer in at most two Chinese sentences.")
      evidence: list[str] = Field(
          description="Two concise supporting points.", min_length=2, max_length=2
      )
      limitation: str = Field(description="One uncertainty or scope limitation.")
  
  
  class FinalAnswer(BaseModel):
      """The public response contract returned by the coordinator."""
  
      answer: str = Field(description="A concise Chinese answer for the user.")
      key_points: list[str] = Field(
          description="Exactly three user-facing key points.", min_length=3, max_length=3
      )
  
  
  class UserInput(TypedDict):
      """Public graph input: only information the model needs to answer."""
  
      user_profile: str
      user_question: str
  
  
  class CoordinatorState(UserInput):
      plan: ResearchPlan
      findings: list[Finding]
      final_answer: FinalAnswer
  
  
  class ResearcherInput(TypedDict):
      """Explicit projection from coordinator state into a researcher subgraph."""
  
      user_profile: str
      user_question: str
      task: Subtask
  
  
  class ResearcherState(ResearcherInput):
      finding: Finding
  
  
  class ResearcherOutput(TypedDict):
      finding: Finding
  
  
  class CoordinatorOutput(TypedDict):
      final_answer: FinalAnswer
  
  
  configurable_model = init_chat_model(
      configurable_fields=("model", "max_tokens", "api_key"),
  )
  
  
  def model_for(settings: Configuration, max_tokens: int):
      return configurable_model.with_config(
          {
              "configurable": {
                  "model": settings.research_model,
                  "max_tokens": max_tokens,
              },
              "tags": ["langsmith:nostream", "learning:multi-agent-handoff"],
          }
      )
  
  
  async def research_one(
      state: ResearcherState,
      runtime: Runtime[Configuration],
  ) -> ResearcherOutput:
      task = state["task"]
      finding = await model_for(runtime.context, 160).with_structured_output(
          Finding
      ).ainvoke(
          [
              HumanMessage(
                  content=(
                      "你是研究子 agent。只完成分配给你的一个任务，不要重新规划。\n"
                      f"用户画像: {state['user_profile']}\n"
                      f"用户问题: {state['user_question']}\n"
                      f"任务主题: {task.topic}\n"
                      f"任务问题: {task.question}\n"
                      f"要求证据: {task.expected_evidence}"
                  )
              )
          ]
      )
      return {"finding": finding}
  
  
  researcher_graph = (
      StateGraph(
          ResearcherState,
          context_schema=Configuration,
          input_schema=ResearcherInput,
          output_schema=ResearcherOutput,
      )
      .add_node("research_one", research_one)
      .add_edge(START, "research_one")
      .add_edge("research_one", END)
      .compile()
  )
  
  
  async def make_plan(
      state: CoordinatorState,
      runtime: Runtime[Configuration],
  ):
      plan = await model_for(runtime.context, 180).with_structured_output(
          ResearchPlan
      ).ainvoke(
          [
              HumanMessage(
                  content=(
                      "你是主协调 agent。把用户问题拆成两个互补、可并行的研究任务。\n"
                      f"用户画像: {state['user_profile']}\n"
                      f"用户问题: {state['user_question']}"
                  )
              )
          ]
      )
      return {"plan": plan}
  
  
  async def run_researchers(
      state: CoordinatorState,
      runtime: Runtime[Configuration],
  ):
      results = await asyncio.gather(
          *(
              researcher_graph.ainvoke(
                  {
                      "user_profile": state["user_profile"],
                      "user_question": state["user_question"],
                      "task": task,
                  },
                  context=runtime.context,
              )
              for task in state["plan"].tasks
          )
      )
      return {"findings": [result["finding"] for result in results]}
  
  
  async def write_answer(
      state: CoordinatorState,
      runtime: Runtime[Configuration],
  ) -> CoordinatorOutput:
      findings = "\n".join(
          finding.model_dump_json() for finding in state["findings"]
      )
      final_answer = await model_for(runtime.context, 220).with_structured_output(
          FinalAnswer
      ).ainvoke(
          [
              HumanMessage(
                  content=(
                      "你是主协调 agent。仅依据下列子 agent 的结构化 finding 回答用户；"
                      "不要编造未提供的证据，并保留必要限制。\n"
                      f"用户画像: {state['user_profile']}\n"
                      f"用户问题: {state['user_question']}\n"
                      f"研究计划: {state['plan'].model_dump_json()}\n"
                      f"子 agent 结果:\n{findings}"
                  )
              )
          ]
      )
      return {"final_answer": final_answer}
  
  
  coordinator_graph = (
      StateGraph(
          CoordinatorState,
          context_schema=Configuration,
          input_schema=UserInput,
          output_schema=CoordinatorOutput,
      )
      .add_node("make_plan", make_plan)
      .add_node("run_researchers", run_researchers)
      .add_node("write_answer", write_answer)
      .add_edge(START, "make_plan")
      .add_edge("make_plan", "run_researchers")
      .add_edge("run_researchers", "write_answer")
      .add_edge("write_answer", END)
      .compile()
  )
  
  
  async def main():
      result = await coordinator_graph.ainvoke(
          {
              "user_profile": "Python 初学者，正在学习当前 open_deep_research 项目。",
              "user_question": "主 agent 和 researcher 子 agent 怎样通过结构化数据协作？",
          },
          context=Configuration.from_env(),
      )
      answer = result["final_answer"]
      print("最终回答: " + answer.answer)
      print("要点: " + " | ".join(answer.key_points))
  
  
  if __name__ == "__main__":
      asyncio.run(main())