- query生成
class SearchQueryList(BaseModel):
query: List[str] = Field(
description="A list of search queries to be used for web research."
)
rationale: str = Field(
description="A brief explanation of why these queries are relevant to the research topic."
)生成 若干 query + rationale
prompt 如下 →
query_writer_instructions = """Your goal is to generate sophisticated and diverse web search queries. These queries are intended for an advanced automated web research tool capable of analyzing complex results, following links, and synthesizing information.
Instructions:
- Always prefer a single search query, only add another query if the original question requests multiple aspects or elements and one query is not enough.
- Each query should focus on one specific aspect of the original question.
- Don't produce more than {number_queries} queries.
- Queries should be diverse, if the topic is broad, generate more than 1 query.
- Don't generate multiple similar queries, 1 is enough.
- Query should ensure that the most current information is gathered. The current date is {current_date}.
Format:
- Format your response as a JSON object with ALL three of these exact keys:
- "rationale": Brief explanation of why these queries are relevant
- "query": A list of search queries
Example:
Topic: What revenue grew more last year apple stock or the number of people buying an iphone
```json
{{
"rationale": "To answer this comparative growth question accurately, we need specific data points on Apple's stock performance and iPhone sales metrics. These queries target the precise financial information needed: company revenue trends, product-specific unit sales figures, and stock price movement over the same fiscal period for direct comparison.",
"query": ["Apple total revenue growth fiscal year 2024", "iPhone unit sales growth fiscal year 2024", "Apple stock price growth fiscal year 2024"],
}}Context: {research_topic}"""
根据context 这里就是聊天历史 生成query (需彼此不相同 只有topic很广生成多个 搜索是加入现在时间)
最后只取query
1. 创建multiple 搜索的node 1对1分配每个query
这个是之前不常用的 可以用 conditional edge + send 使一个上游node对应多个下游node 每个分配不同state
map reduce 工作流
```python
def continue_to_web_research(state: QueryGenerationState):
"""LangGraph node that sends the search queries to the web research node.
This is used to spawn n number of web research nodes, one for each search query.
"""
return [
Send("web_research", {"search_query": search_query, "id": int(idx)})
for idx, search_query in enumerate(state["query_list"])
]builder.add_conditional_edges(
"generate_query", continue_to_web_research, ["web_research"]
)
#同时注意这里 在send时 根据不同node所需state 自动变化需要传进去的数值
# 比如只提供给websearch node 简单的state 而且是从上一个node 返回值就确定好的
#一个例子是这里 query创建的返回值是 return {"query_list": result.query}
# 是下一个node(甚至只是add_conditional_edges中的中转 但是还是变成了所需state)
class OverallState(TypedDict):
messages: Annotated[list, add_messages]
search_query: Annotated[list, operator.add]
web_research_result: Annotated[list, operator.add]
sources_gathered: Annotated[list, operator.add]
initial_search_query_count: int
max_research_loops: int
research_loop_count: int
reasoning_model: str
class WebSearchState(TypedDict):
search_query: str
id: str
class QueryGenerationState(TypedDict):
query_list: list[Query]- Web search
web_searcher_instructions = """Conduct targeted Google Searches to gather the most recent, credible information on "{research_topic}" and synthesize it into a verifiable text artifact.
Instructions:
- Query should ensure that the most current information is gathered. The current date is {current_date}.
- Conduct multiple, diverse searches to gather comprehensive information.
- Consolidate key findings while meticulously tracking the source(s) for each specific piece of information.
- The output should be a well-written summary or report based on your search findings.
- Only include the information found in the search results, don't make up any information.
Research Topic:
{research_topic}
"""from google.genai import Client
genai_client = Client(api_key=os.getenv("GEMINI_API_KEY"))
# Uses the google genai client as the langchain client doesn't return grounding metadata
response = genai_client.models.generate_content(
model=configurable.query_generator_model,
contents=formatted_prompt,
config={
"tools": [{"google_search": {}}],
"temperature": 0,
},
)# resolve the urls to short urls for saving tokens and time
# 注意这里的id 是从上面query的id传下来的
resolved_urls = resolve_urls(
response.candidates[0].grounding_metadata.grounding_chunks, state["id"]
)
# Gets the citations and adds them to the generated text
#给每个url 独立id
citations = get_citations(response, resolved_urls)
modified_text = insert_citation_markers(response.text, citations)
sources_gathered = [item for citation in citations for item in citation["segments"]]
return {
"sources_gathered": sources_gathered,
"search_query": [state["search_query"]],
"web_research_result": [modified_text],
}#注意这里返回的时候把search query放到list中 下一步得以汇总query和result
return {
"sources_gathered": sources_gathered,
"search_query": [state["search_query"]],
"web_research_result": [modified_text],
}- reflection
reflection_instructions = """You are an expert research assistant analyzing summaries about "{research_topic}".
Instructions:
- Identify knowledge gaps or areas that need deeper exploration and generate a follow-up query. (1 or multiple).
- If provided summaries are sufficient to answer the user's question, don't generate a follow-up query.
- If there is a knowledge gap, generate a follow-up query that would help expand your understanding.
- Focus on technical details, implementation specifics, or emerging trends that weren't fully covered.
Requirements:
- Ensure the follow-up query is self-contained and includes necessary context for web search.
Output Format:
- Format your response as a JSON object with these exact keys:
- "is_sufficient": true or false
- "knowledge_gap": Describe what information is missing or needs clarification
- "follow_up_queries": Write a specific question to address this gap
Example:
```json
{{
"is_sufficient": true, // or false
"knowledge_gap": "The summary lacks information about performance metrics and benchmarks", // "" if is_sufficient is true
"follow_up_queries": ["What are typical performance benchmarks and metrics used to evaluate [specific technology]?"] // [] if is_sufficient is true
}}Reflect carefully on the Summaries to identify knowledge gaps and produce a follow-up query. Then, produce your output following this JSON format:
Summaries: {summaries} """
```python
class Reflection(BaseModel):
is_sufficient: bool = Field(
description="Whether the provided summaries are sufficient to answer the user's question."
)
knowledge_gap: str = Field(
description="A description of what information is missing or needs clarification."
)
follow_up_queries: List[str] = Field(
description="A list of follow-up queries to address the knowledge gap."
)
# Increment the research loop count and get the reasoning model
state["research_loop_count"] = state.get("research_loop_count", 0) + 1
reasoning_model = state.get("reasoning_model") or configurable.reasoning_model- 若抓取到的信息 not sufficient → 回归web search 直到reflection次数到上限
builder.add_conditional_edges(
"reflection", evaluate_research, ["web_research", "finalize_answer"]
)if state["is_sufficient"] or state["research_loop_count"] >= max_research_loops:
return "finalize_answer"
else:
return [
Send(
"web_research",
{
"search_query": follow_up_query,
"id": state["number_of_ran_queries"] + int(idx),
},
)
for idx, follow_up_query in enumerate(state["follow_up_queries"])
]- 准备答案
answer_instructions = """Generate a high-quality answer to the user's question based on the provided summaries.
Instructions:
- The current date is {current_date}.
- You are the final step of a multi-step research process, don't mention that you are the final step.
- You have access to all the information gathered from the previous steps.
- You have access to the user's question.
- Generate a high-quality answer to the user's question based on the provided summaries and the user's question.
- you MUST include all the citations from the summaries in the answer correctly.
User Context:
- {research_topic}
Summaries:
{summaries}"""注意点1:
builder = StateGraph(OverallState, config_schema=Configuration)
可以把 config 独立于 state pass 到每一个node中
def web_research(state: WebSearchState, config: RunnableConfig) -> OverallState: