解开复杂的人工智能任务：使用 Gemini 2.0、LangGraph 和 Grounded Responses 的多步骤代理

Rifx.Online
Programming , Machine Learning , Autonomous Systems
19 Dec, 2024

如何构建不仅理解复杂请求而且能够执行必要步骤以满足这些请求的 AI？

本文探讨了一种强大的解决方案：由 Gemini 2.0 和 LangGraph 框架驱动的多步骤 AI 代理。这些代理协调复杂的工作流程，并通过基于实际数据的响应增强其推理能力，从 Google 搜索、BigQuery 和第三方 API 获取真实世界的数据。

今日的挑战

第一个重大突破是 LLM。想象一下一个充满大量信息的“大脑”，能够理解和生成自然语言。但是，这个“脑”有一个根本的局限性：它的知识仅限于其训练数据——这是特定时间点的信息快照。

然后出现了 RAG。这将“大脑”连接到配置的数据源，使 LLM 能够访问专有或外部知识，实际上使它们能够接触到互联网和特定知识库的广泛且不断更新的资源。这使得 LLM 能够提供更相关、更准确和更及时的答案，显著增强了它们的实用性。

然而，即使有了 RAG，人工智能在处理复杂的多步骤任务时仍然存在困难，这些任务需要与现实世界的互动。想象一下请求帮助策划一个惊喜派对：一个增强了 RAG 的 LLM 可以建议主题和场地，但无法预订场地或发送邀请函。

在金融等领域，这种局限性更为关键。一个人类金融分析师在建议投资组合再平衡时会系统地：1）评估当前持有的资产和市场价值，2）根据市场趋势分析增长潜力，3）推荐买入/卖出决策。这种多步骤推理超出了标准的 RAG 增强 LLM 的能力。

代理型人工智能：从知识到行动

今天，我们将探讨如何通过代理型人工智能克服这些限制。代理型人工智能将大型语言模型（LLM）从一个被动的“脑”转变为一个主动的、智能的代理，可以制定计划，利用工具与外部系统（API、数据库）进行交互，收集数据，执行计算，并执行任务以实现特定目标。

代理不仅仅是提供信息，而是行动以实现期望的结果。

复制人类推理：股票投资组合案例

本次探索专注于使用Agentic AI进行一个实际案例：管理个人股票投资组合。目标是创建一个能够处理复杂查询的代理，例如：“考虑到当前市场和价格，我应该卖掉我的特斯拉股票吗？”

人类交易者通常会：

检查他们的特斯拉持股和平均购买价格。
检查特斯拉股票的当前市场价格。
研究特斯拉的市场前景和相关行业趋势。
根据这一分析，决定是持有、再买入还是卖出。

我们将探讨是否可以使用Agentic AI复制这种多步骤推理和行动。

演示

请查看 Agentic AI 在下面的演示中处理复杂金融查询。为了快速了解编排，我们建议以 2 倍速观看。

解决方案详情

在这个解决方案中，我使用了一种“计划-执行”风格的代理，灵感来自于LangChain的计划-执行教程

该代理的关键组成部分包括：

Gemini 2.0 (LLM)： 负责理解用户请求、制定查询和综合信息的“大脑”。
LangGraph (编排)： 管理工作流程的框架，将复杂任务分解为顺序步骤，并在组件之间路由数据。
基础数据源： 来自Google搜索（用于市场趋势和新闻）、BigQuery（用于用户投资组合详情）和FinnHub API（用于当前股价）的实时信息。

工作流程如下进行：

(1) 制定计划并生成任务：

当用户提出问题时，LangGraph将工作流程路由到计划代理。该代理由Gemini 2.0提供支持，使用精心设计的提示生成一个结构化为Plan对象的逐步计划：

class Plan(BaseModel):
    steps: List[str] = Field(description="Steps to follow, in order.")

这个Plan对象随后在PlanExecute TypedDict中使用，以管理工作流程状态：

class PlanExecute(TypedDict):
    input: str
    plan: List[str]
    past_steps: Annotated[List[Tuple[str, str]], operator.add]
    response: Optional[str]
    intermediate_responses: List[str]

计划者的提示引导Gemini创建一个清晰、简明的计划，为每一步选择合适的工具。

PLANNER_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """You are an expert in deciphering questions and creating step-by-step plans.
Based on the given objective, create a simple plan. Each step should be a distinct task that, when executed, will lead to the correct answer. Avoid superfluous steps.
Use these guidelines to choose the right tool:
- Portfolio retrieval (e.g., "What's my portfolio?", "What are my holdings?", "Last trade on Nvidia?"): Use the {{portfolio_retriever}} tool.
- Check the current price of a stock using the stock symbol(e.g. current price of GOOG): Use the {{price_checker}} tool. If there are multiple stocks to check, break down into multiple steps to do multiple function calls to check current price for individual stock.
- Equity/market analysis (e.g., "Will Nvidia rise?", "Current stock price?", "Is Intel a buy?", "What are the risks?"): Use the {{stock_analyser}} tool.
- General/non-financial questions (e.g., "Hi", "Who are you?"): Use the {{normal_responder}} tool.
The final step's result should be the final answer. Ensure each step has enough information; do not skip steps.
""",
        ),
        ("placeholder", "{messages}"),
    ]
)

(2) 执行任务和重新规划：

一旦计划者生成了逐步计划，LangGraph将其传递给代理执行器进行执行。在此阶段，代理根据计划中的每一步利用适当的工具。

例如，步骤1需要检索投资组合信息，代理使用portfolio_retriever工具。该工具将自然语言指令转换为SQL查询，并在BigQuery数据集上执行该查询（用户持有的股票存储在该数据集中）。

每个步骤执行后，工作流程将路由到重新规划器。重新规划器的角色至关重要：它综合到目前为止收集的信息，并确定下一步行动。

replan_step函数协调此过程：

async def replan_step(state: PlanExecute):
    all_responses = "\n".join(state["intermediate_responses"])
    all_steps = "\n".join([f"{step}: {response}" for step, response in state["past_steps"]])
    context = f"Here is the information gathered from the previous steps:\n{all_steps}\n\nHere are the direct responses from the tools:\n{all_responses}"

    output = await replanner.ainvoke({**state, "input": context})
    if output.response:
        cleaned_response = clean_newlines(output.response.response)
        with cl.Step(name="Final Response"):
            await cl.Message(content="**Final Response:**").send()
            await cl.Message(content=cleaned_response).send()
        return {"response": cleaned_response}
    else:
        return {"plan": output.plan.steps}

重新规划器接收一个context，包含：

已完成步骤及其对应响应的摘要。
从工具直接收到的原始输出。

利用这个上下文，重新规划器（由Gemini 2.0和特定提示提供支持）决定是否有足够的信息来提供最终答案。

最终响应： 如果重新规划器生成了最终响应（output.response存在），则该响应发送给用户，工作流程结束。
继续执行： 如果未生成最终响应，重新规划器将现有计划传回代理执行器，后者将执行下一步。

这一执行和重新规划的循环持续进行，直到重新规划器确定用户的查询已完全解决。

此逻辑在LangGraph中使用这些边缘实现：

def should_end(state: PlanExecute):
    return END if "response" in state and state["response"] is not None else "agent"

## --- 工作流程定义 ---
workflow = StateGraph(PlanExecute)
workflow.add_node("planner", plan_step)
workflow.add_node("agent", execute_step)
workflow.add_node("replan", replan_step)
workflow.add_edge(START, "planner")
workflow.add_edge("planner", "agent")
workflow.add_edge("agent", "replan")
workflow.add_conditional_edges("replan", should_end, {"agent": "agent", END: END})
app = workflow.compile()

workflow.add_edge("agent", "replan")行创建了从代理执行器到重新规划器的直接路径。

workflow.add_conditional_edges行定义了重新规划步骤后发生的逻辑。

如果should_end为真，工作流程转到END状态。
如果should_end为假，工作流程返回到agent状态（代理执行器）。

(3) 继续执行

在检索投资组合信息后，工作流程返回到代理执行器，以执行计划中的下一步：检查特斯拉股票的当前市场价格。代理执行器使用price_checker工具，查询FinnHub API以获取实时股价。

然后，工作流程返回到重新规划器。由于需要更多信息来回答初始查询，重新规划器将工作流程指向代理执行器。

这次，代理执行器执行涉及分析特斯拉市场表现并确定是否建议持有或出售的步骤。为此，代理使用stock_analyser工具，该工具利用Google搜索基础提供最新和相关的信息。

stock_analyser工具使用以下函数通过Google搜索结果进行分析。这一函数向Google搜索基础服务发送提示，指示其查找与正在分析的股票相关的新闻和信息。这确保分析基于最新的信息。

def google_ground(prompt: str) -> str:
    request = discoveryengine.GenerateGroundedContentRequest(
        location=google_search_client.common_location_path(
                project=PROJECT_NUMBER, location="global"
        ),
        generation_spec=spec,
        contents=[
            discoveryengine.GroundedGenerationContent(
                    role="user",
                    parts=[discoveryengine.GroundedGenerationContent.Part(text=prompt)],
            )
        ],
        system_instruction=discoveryengine.GroundedGenerationContent(
            parts=[
                discoveryengine.GroundedGenerationContent.Part(text="If given a stock or option to analyse, try to find relevant news from google search to see if it's a good time to buy more or sell. Use these news to formulate your analysis and return a comprehensive response targted to the provided stock or option. Rmb that you are a seasoned investment analyst so always remove any disclaimers about this not being financial advice.")
            ],
        ),
        grounding_spec=discoveryengine.GenerateGroundedContentRequest.GroundingSpec(
            grounding_sources=[
                discoveryengine.GenerateGroundedContentRequest.GroundingSource(
                        google_search_source=discoveryengine.GenerateGroundedContentRequest.GroundingSource.GoogleSearchSource()
                )
            ]
        ),
    )
    google_responses = google_search_client.generate_grounded_content(request)
    
    return_prompt=f"""Generate a natural language response based on the original question: '{prompt}' and the returned results: '{google_responses}'"""
    response=model.generate_content(return_prompt)
    return response.text