AI Agent

发表于 | 分类于 | 阅读次数

目前，与AI交互遵循一种熟悉的流程。您都必须输入prompt，AI 模型会根据输入来响应。每次您想要新的输出时，您都必须提供prompt。总是有人来启动这个过程。

AI agent以不同的方式工作。他们被设计为独立思考和行动。您唯一需要提供的就是一个目标。他们将根据环境的反馈和自己的内心独白生成一个任务列表并开始工作。就好像AI Agent可以自我提示，不断发展和适应，可以在存在大量新信息的不可预测的环境中工作，以尽可能最好的方式实现他们的目标。

AutoGPT

https://aiedge.medium.com/autogpt-forge-e3de53cc58ec

img

Limitations

  • Wrong tool selection
  • Infinite loops 无限循环
  • Hallucinations

Profile

当我们人类专注于各种任务时，我们会为这些任务调整自己的状态。无论是写作、切菜、驾驶还是进行体育运动，我们都会集中注意力，甚至采取不同的心态。当讨论Agent时，概念上的profile指的就是这种适应性。研究表明，仅仅告知一个Agent程序它在某个特定任务上是专家，就能提高它的性能。

profile模块具有超越仅仅优化提示的潜在应用。它可以用于调整Agent程序的记忆功能、可用动作，甚至是驱动Agent程序的底层大型语言模型（LLM）。

Memory

对于一个机器人来说，记忆不仅仅是存储，它是构建其身份、能力和学习的基础。正如我们的记忆影响我们的决策、反应甚至个性一样，机器人的记忆是其过去互动、学习和反馈的积累。记忆主要分为长期记忆和短期记忆。

  • 长期记忆类似于机器人的基础知识，为 Agent 提供长时间保留和回忆信息的能力，这个时候需要借助外部问量存储和快速检索来实现(向量数据库)

  • 短期记忆（或工作记忆）关注的是即时的事务，处理短暂记忆，就像我们对最近事件的回忆一样。虽然对于实时任务至关重要，但并不是所有的短期记忆都能进入机器人的长期存储。（Prompt Engineering，上下文）

在这个领域出现了一个新兴的概念，即记忆反思。在这里，机器人不仅仅是存储记忆，还会主动回顾它们。这种内省使机器人能够重新评估、优先处理或甚至丢弃信息，就像人类回忆和从过去经验中学习一样。

Planning

规划是机器人解决问题的路线图。当面对复杂的挑战时，人类本能地将其分解为可管理的小任务，这种策略也被镜像在基于LLM的机器人中。这种有条不紊的方法使机器人能够以结构化的思维方式解决问题，确保全面而系统的解决方案。

机器人的规划工具包中有两种主要策略。

  • 第一种是带反馈的规划，这是一种自适应的方法。在这种方法中，机器人根据结果来优化其策略，就像根据用户反馈不断迭代设计版本一样。

  • 第二种是不带反馈的规划，将机器人视为一名策略家，仅依靠其现有知识和远见。将复杂任务分解为更小、更易于处理的子目标,从而实现对复杂任务的高效处理。

Action

在回忆和规划之后，最终到来的是行动。这是Agent认知过程转化为实际结果的阶段，运用Agent的能力。每个决策、每个思考都在行动阶段得到体现，将抽象概念转化为明确的成果。

无论是写下回应、保存文件还是启动新的流程，行动是Agent决策之旅的关键。它连接着数字认知和真实世界的影响，将Agent的电子冲动转化为有意义而有目标的结果。

AutoGen

AutoGen 是一个框架，可以使用多个代理程序进行交流来解决任务，实现 LLM 应用的开发。AutoGen 代理程序可定制，可对话，并且能够无缝地与人类参与结合起来。它们可以以不同的方式运行，结合了 LLM、人类输入和工具的各种组合模式。

AutoGen Overview

AutoGen 可以通过多个代理程序之间的对话，以最小的努力构建下一代 LLM 应用。它简化了复杂的 LLM 工作流程的编排、自动化和优化。它最大限度地提高了 LLM 模型的性能，并克服了它们的不足之处。 它支持各种复杂工作流程的多种对话模式。借助可定制和可对话的代理程序，开发人员可以使用 AutoGen 构建各种不同的对话模式，包括对话自治性、代理程序数量和代理程序对话拓扑结构等。 它提供了一系列具有不同复杂度的工作系统。这些系统涵盖了各种不同领域和复杂度的应用。它们展示了 AutoGen 如何轻松支持不同的对话模式。 AutoGen 提供了 openai.Completion 或 openai.ChatCompletion 的即插即用替代方案，作为增强型推理 API。它可以轻松进行性能调优，提供了 API 统一化和缓存等实用工具，支持更高级的用法模式，例如错误处理、多配置推理、上下文编程等。

总结

Agent = LLM + 计划+执行+纠错

Agent

  • 分解任务并完成 (ToT,Reasoning 推理 + Action 行动 React)
  • 历史的动作进行自我反思并完善 (通过长期记忆 ，Reflexion 反思)

相比RAG，但可以调用更加多的工具或者权限（例如实时信息）去完成更多实际任务而不仅仅是调用向量数据库输出结果。

image-20240327213817219

将Agent视为在RAG顶部包裹一层的东西，Agent动态地丰富查询信息，基本上允许这种整体上的更高级抽象以正确的方式使用工具，试图为您提供响应

image-20240327214416854

ToT

image-20240330170435602

  1. Thought: 只有一个中间计划
  2. 生成思路：样例
  3. 评估思路：投票
  4. 搜索算法：宽度优先搜索（深度=2，广度=5）

multi-agent

解决sigle agent遇到的问题

sigle multi 备注
如何完成需要不同背景的复杂任务 多agent根据标准或自定义流程配合 流程可能复杂且多样，增加编程难度
如何提高应用的可靠性 多个agent讨论、复盘逻辑 依然建立在LLM能服从指令的前提下
如何灵活兼容多模态数据 招募不同擅长领域的agent合作 如何高效保存、分享多模态数据
如何提高解决问题的效率 优化、并行多个子任务agent执行 如何并行，可以自动化优化吗
  • 如何编排复杂流程(灵活、交流机制)：交流顺序、方式复杂多变，逐一枚举费时费力
  • 如何提高鲁棒性和可靠性：大模型幻觉和不稳定的指令跟随能力会影响应用运行效果
  • 如何处理多模态数据：需要在文本支持的基础上兼容多模态数据的传递，存储和展示
  • 如何提高运行效率：:需要分布式背景，并深度分析应用流程，对开发者而言优化难度高

langchain 参考

crewai

先思考两个问题

  • 每个agent的功能
  • agent之间怎样连接

An agent supervisor 路由到 individual agents.

img

多agent，互相协助

img

分层代理团队 , 上面两个的结合，多个graph 连接

img

Plan-and-Execute Agents

参考

plan-and-execute agents 对比 Reasoning and Action (ReAct)-style agents会更好，目前还没完美实现。

  • 通过强制规划器明确“思考”完成整个任务所需的所有步骤，它们可以在整体上表现更好（任务完成率和质量方面）。生成完整的推理步骤是一种改进结果的可靠提示技术。将问题细分还可以实现更有针对性的任务执行。
  • 可以分散并行地做，效率更高
  • token费用更少，每一步也不一定依赖调用大模型，只有(re-)planning steps and to generate the final response才需要。

总结

  • ReAct处理复杂问题时候： 流程长，有Long term Memory 需要额外存储，看LLM性能（token多），时间长，效果不一定好。
  • Plan-and-Execute 对于planner 要求高，任务执行管理麻烦

一般分成3块

  • 计划制定：LLM生成文本，直接回应用户或传递给函数。
  • 执行：你的代码调用其他软件执行操作，比如查询数据库或调用API。
  • 分析：根据工具调用的响应做出反应，要么调用另一个函数，要么回应用户。

image-20240331153650462

Plan-and-Execute 实现

核心思想是首先制定一个多步计划，然后逐个执行计划中的任务。在完成特定任务之后，您可以重新审视计划并根据需要进行修改。

  • 只能串行，不能产生并行计划，需要计划编排生成技巧，加快执行速度

paper

Reasoning WithOut Observations 实现

可以避免在每个任务中都需要使用一个LLM（语言模型）的问题，同时允许任务依赖于前一个任务的结果。这可以通过在规划器的输出中允许变量赋值来实现。下面是代理系统设计的示意图。ReWOO

image-20240331160444661

根据问题，规划器在工具响应之前组合了一份全面的互联计划蓝图。该蓝图指导worker使用外部工具并收集证据。最后，计划和证据被配对并传递给求解器以获取答案。

  • 关键在于任务，列出需要的Evidence，有标准输出
  • 还是得串行

LLMCompiler

paper

image-20240331161915159

  • executor可以让 tasks 并行
  • planner 流式输出
  • solver 有replan功能

img

Reflection Agents

参考

Basic Reflection

  • generator ：尝试直接响应用户的请求。
  • reflector：提示扮演教师的角色，并对最初的反应提出建设性的批评。

循环进行固定次数，并返回最终生成的输出。

由于反射步骤不基于任何外部过程，因此最终结果可能不会比原始结果好得多。让我们探索一些可以改善这种情况的其他技术。

image-20240331165845052

Reflexion

在反思中，行动者智能体明确批评每个响应，并将其批评建立在外部数据的基础上。它被迫生成引用并明确列举生成的响应中多余和缺失的方面。这使得反思的内容更具建设性，并更好地引导生成器响应反馈。

它只追求一个固定的轨迹，所以如果它犯了一个错误，这个错误可能会影响后续的决策。

img

paper 实现

它结合了反射/评估和搜索（特别是蒙特卡罗树搜索），与 ReACT、Reflexion 甚至 Tree of Thoughts 等类似技术相比，可以实现更好的整体任务性能。它采用标准的强化学习 (RL) 任务框架，通过调用 LLM 来替换 RL 代理、价值函数和优化器。这旨在帮助代理适应复杂任务并解决问题，避免陷入重复循环。

img

  1. 选择：根据下面步骤 (2) 中的总奖励选择最佳的下一步行动。要么做出响应（如果找到解决方案或达到最大搜索深度），要么继续搜索。
  2. 扩展和模拟：生成 N（在我们的例子中为 5）个潜在操作以并行执行并执行它们。
  3. 反思+评估：观察这些行动的结果，并根据反思（以及可能的外部反馈）对决策进行评分
  4. 反向传播：根据结果更新根轨迹的分数。

img

LlamaIndex

router-query-engine example

1
2
3
4
5
6
7
8
9
10
11
12
from llama_index.core.query_engine.router_query_engine import RouterQueryEngine
from llama_index.core.selectors import LLMSingleSelector

# 路由
query_engine = RouterQueryEngine(
    selector=LLMSingleSelector.from_defaults(),
    query_engine_tools=[
        summary_tool,
        vector_tool,
    ],
    verbose=True
)

tool calling

Define a Simple Tool

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from llama_index.core.tools import FunctionTool

def add(x: int, y: int) -> int:
    """Adds two integers together."""
    return x + y

def mystery(x: int, y: int) -> int: 
    """Mystery function that operates on top of two numbers."""
    return (x + y) * (x + y)


add_tool = FunctionTool.from_defaults(fn=add)
mystery_tool = FunctionTool.from_defaults(fn=mystery)


# 调用function tool
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-3.5-turbo")
response = llm.predict_and_call(
    [add_tool, mystery_tool], 
    "Tell me the output of the mystery function on 2 and 9", 
    verbose=True
)
print(str(response))

Define the Auto-Retrieval Tool

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from typing import List
from llama_index.core.vector_stores import FilterCondition


def vector_query(
    query: str, 
    page_numbers: List[str]
) -> str:
    """Perform a vector search over an index.
    
    query (str): the string query to be embedded.
    page_numbers (List[str]): Filter by set of pages. Leave BLANK if we want to perform a vector search
        over all pages. Otherwise, filter by the set of specified pages.
    
    """

    metadata_dicts = [
        {"key": "page_label", "value": p} for p in page_numbers
    ]
    
    query_engine = vector_index.as_query_engine(
        similarity_top_k=2,
        filters=MetadataFilters.from_dicts(
            metadata_dicts,
            condition=FilterCondition.OR
        )
    )
    response = query_engine.query(query)
    return response
    

vector_query_tool = FunctionTool.from_defaults(
    name="vector_tool",
    fn=vector_query
)

Setup Function Calling Agent

image-20240510215117242

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from llama_index.core.agent import FunctionCallingAgentWorker
from llama_index.core.agent import AgentRunner

agent_worker = FunctionCallingAgentWorker.from_tools(
    [vector_tool, summary_tool], 
    llm=llm, 
    verbose=True
)
agent = AgentRunner(agent_worker)
response = agent.query(
    "Tell me about the agent roles in MetaGPT, "
    "and then how they communicate with each other."
)


response = agent.chat(
    "Tell me about the evaluation datasets used."
)

task debug

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
agent_worker = FunctionCallingAgentWorker.from_tools(
    [vector_tool, summary_tool], 
    llm=llm, 
    verbose=True
)
agent = AgentRunner(agent_worker)

task = agent.create_task(
    "Tell me about the agent roles in MetaGPT, "
    "and then how they communicate with each other."
)

step_output = agent.run_step(task.task_id)

completed_steps = agent.get_completed_steps(task.task_id)
print(f"Num completed for task {task.task_id}: {len(completed_steps)}")
print(completed_steps[0].output.sources[0].raw_output)


upcoming_steps = agent.get_upcoming_steps(task.task_id)
print(f"Num upcoming steps for task {task.task_id}: {len(upcoming_steps)}")
upcoming_steps[0]


step_output = agent.run_step(
    task.task_id, input="What about how agents share information?"
)
step_output = agent.run_step(task.task_id)
print(step_output.is_last)
response = agent.finalize_response(task.task_id)

graph agent example

https://langchain-ai.github.io/langgraph/tutorials/customer-support/customer-support/

Part 1 Diagram

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

builder = StateGraph(State)


# Define nodes: these do the work
builder.add_node("assistant", Assistant(part_1_assistant_runnable))
builder.add_node("action", create_tool_node_with_fallback(part_1_tools))
# Define edges: these determine how the control flow moves
builder.set_entry_point("assistant")
builder.add_conditional_edges(
    "assistant",
    tools_condition,
    # "action" calls one of our tools. END causes the graph to terminate (and respond to the user)
    {"action": "action", END: END},
)
builder.add_edge("action", "assistant")

# checkpoint 保存持久话
# The checkpointer lets the graph persist its state
# this is a complete memory for the entire graph.
memory = SqliteSaver.from_conn_string(":memory:")
part_1_graph = builder.compile(checkpointer=memory)

打印图

1
2
3
4
5
6
7
from IPython.display import Image, display

try:
    display(Image(part_1_graph.get_graph(xray=True).draw_mermaid_png()))
except:
    # This requires some extra dependencies and is optional
    pass

session 管理 同一个人 不同 session

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import shutil
import uuid

# Let's create an example conversation a user might have with the assistant
tutorial_questions = [
    "Hi there, what time is my flight?",
    .....
    "OK great pick one and book it for my second day there.",
]

# Update with the backup file so we can restart from the original place in each section
shutil.copy(backup_file, db)
thread_id = str(uuid.uuid4())

config = {
    "configurable": {
        # The passenger_id is used in our flight tools to
        # fetch the user's flight information
        "passenger_id": "3442 587242",
        # Checkpoints are accessed by thread_id   session 管理 同一个人 不同 session
        "thread_id": thread_id,
    }
}


_printed = set()
for question in tutorial_questions:
    events = part_1_graph.stream(
        {"messages": ("user", question)}, config, stream_mode="values"
    )
    for event in events:
        _print_event(event, _printed)

Part 2 diagram

需要用户去确认

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    # 保留用户信息
    user_info: str
    
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

builder = StateGraph(State)


def user_info(state: State):
    return {"user_info": fetch_user_flight_information.invoke({})}


# NEW: The fetch_user_info node runs first, meaning our assistant can see the user's flight information without
# having to take an action
builder.add_node("fetch_user_info", user_info)
builder.set_entry_point("fetch_user_info")
builder.add_node("assistant", Assistant(part_2_assistant_runnable))
builder.add_node("action", create_tool_node_with_fallback(part_2_tools))
builder.add_edge("fetch_user_info", "assistant")
builder.add_conditional_edges(
    "assistant", tools_condition, {"action": "action", END: END}
)
builder.add_edge("action", "assistant")

memory = SqliteSaver.from_conn_string(":memory:")
part_2_graph = builder.compile(
    checkpointer=memory,
    # 在
    # NEW: The graph will always halt before executing the "action" node.
    # The user can approve or reject (or even alter the request) before
    # the assistant continues
    interrupt_before=["action"],
)

我们看到了“广泛”的聊天机器人如何依靠单个提示和 LLM 来处理各种用户意图，让我们走得更远。然而，使用这种方法很难为已知意图创建可预测的出色用户体验。

您的图表可以检测用户意图并选择适当的工作流程或“技能”来满足用户的需求。每个工作流程都可以专注于其领域，允许单独的改进，而不会降低整体助手的性能。

在本节中，我们将把用户体验分成单独的子图，形成如下结构：

第 4 部分图

意图的上下文切换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from typing import Annotated, Literal, Optional

from typing_extensions import TypedDict

from langgraph.graph.message import AnyMessage, add_messages


def update_dialog_stack(left: list[str], right: Optional[str]) -> list[str]:
    """Push or pop the state."""
    # 进退stack  出入状态
    if right is None:
        return left
    if right == "pop":
        return left[:-1]
    return left + [right]


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    user_info: str
    dialog_state: Annotated[  # 处于什么状态
        list[
            Literal[
                "assistant",
                "update_flight",
                "book_car_rental",
                "book_hotel",
                "book_excursion",
            ]
        ],
        update_dialog_stack,
    ]

每个步骤都要加入判断任务状态，结束 cancel 与原因信息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class CompleteOrEscalate(BaseModel):
    """A tool to mark the current task as completed and/or to escalate control of the dialog to the main assistant,
    who can re-route the dialog based on the user's needs."""

    cancel: bool = True
    reason: str

    class Config:
        schema_extra = {
            "example": {
                "cancel": True,
                "reason": "User changed their mind about the current task.",
            },
            "example 2": {
                "cancel": True,
                "reason": "I have fully completed the task.",
            },
            "example 3": {
                "cancel": False,
                "reason": "I need to search the user's emails or calendar for more information.",
            },
        }

不同的子图数据格式也不一样

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class ToBookCarRental(BaseModel):
    """Transfers work to a specialized assistant to handle car rental bookings."""

    location: str = Field(
        description="The location where the user wants to rent a car."
    )
    start_date: str = Field(description="The start date of the car rental.")
    end_date: str = Field(description="The end date of the car rental.")
    request: str = Field(
        description="Any additional information or requests from the user regarding the car rental."
    )

    class Config:
        schema_extra = {
            "example": {
                "location": "Basel",
                "start_date": "2023-07-01",
                "end_date": "2023-07-05",
                "request": "I need a compact car with automatic transmission.",
            }
        }


class ToHotelBookingAssistant(BaseModel):
    """Transfer work to a specialized assistant to handle hotel bookings."""

    location: str = Field(
        description="The location where the user wants to book a hotel."
    )
    checkin_date: str = Field(description="The check-in date for the hotel.")
    checkout_date: str = Field(description="The check-out date for the hotel.")
    request: str = Field(
        description="Any additional information or requests from the user regarding the hotel booking."
    )

    class Config:
        schema_extra = {
            "example": {
                "location": "Zurich",
                "checkin_date": "2023-08-15",
                "checkout_date": "2023-08-20",
                "request": "I prefer a hotel near the city center with a room that has a view.",
            }
        }

彻底结束dialog的标记, 或者上下文切换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
# Flight booking assistant
builder.add_node(
    "enter_update_flight",
    create_entry_node("Flight Updates & Booking Assistant", "update_flight"),
)
builder.add_node("update_flight", Assistant(update_flight_runnable))
builder.add_edge("enter_update_flight", "update_flight")
builder.add_node(
    "update_flight_sensitive_tools",
    create_tool_node_with_fallback(update_flight_sensitive_tools),
)
builder.add_node(
    "update_flight_safe_tools",
    create_tool_node_with_fallback(update_flight_safe_tools),
)


def route_update_flight(
    state: State,
) -> Literal[
    "update_flight_sensitive_tools",
    "update_flight_safe_tools",
    "leave_skill",
    "__end__",
]:
    route = tools_condition(state)
    if route == END:
        return END
    tool_calls = state["messages"][-1].tool_calls
    did_cancel = any(tc["name"] == CompleteOrEscalate.__name__ for tc in tool_calls)
    if did_cancel:
        return "leave_skill"
    safe_toolnames = [t.name for t in update_flight_safe_tools]
    if all(tc["name"] in safe_toolnames for tc in tool_calls):
        return "update_flight_safe_tools"
    return "update_flight_sensitive_tools"


builder.add_edge("update_flight_sensitive_tools", "update_flight")
builder.add_edge("update_flight_safe_tools", "update_flight")
builder.add_conditional_edges("update_flight", route_update_flight)


# 彻底退出的标记
# This node will be shared for exiting all specialized assistants
def pop_dialog_state(state: State) -> dict:
    """Pop the dialog stack and return to the main assistant.

    This lets the full graph explicitly track the dialog flow and delegate control
    to specific sub-graphs.
    """
    messages = []
    if state["messages"][-1].tool_calls:
        # Note: Doesn't currently handle the edge case where the llm performs parallel tool calls
        messages.append(
            ToolMessage(
                content="Resuming dialog with the host assistant. Please reflect on the past conversation and assist the user as needed.",
                tool_call_id=state["messages"][-1].tool_calls[0]["id"],
            )
        )
    return {
        "dialog_state": "pop",
        "messages": messages,
    }


builder.add_node("leave_skill", pop_dialog_state)
builder.add_edge("leave_skill", "primary_assistant")