AI Agent 完全指南：从理论到工业实践

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# 传统 LLM：一发，一中，结束
def traditional_llm(prompt: str) -> str:
    return llm.generate(prompt)


# Agent：带状态、带工具、带停止条件的循环
def agent_loop(goal: str) -> str:
    state = init_state(goal)
    for step in range(MAX_STEPS):
        observation = state.latest_observation()
        thought = llm.reason(observation, state.memory)
        action = llm.decide_action(thought, available_tools)
        result = execute(action)
        state.append(thought=thought, action=action, result=result)
        if state.goal_satisfied():
            return state.answer()
    return state.best_effort_answer()

 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 dataclasses import dataclass, field
from typing import Callable, Any


@dataclass
class Tool:
    name: str
    description: str
    schema: dict
    fn: Callable[..., Any]

    def __call__(self, **kwargs) -> dict:
        try:
            return {"ok": True, "result": self.fn(**kwargs)}
        except Exception as exc:
            return {"ok": False, "error": str(exc)}


@dataclass
class WorkingMemory:
    messages: list = field(default_factory=list)
    budget_tokens: int = 4000

    def add(self, role: str, content: str) -> None:
        self.messages.append({"role": role, "content": content})
        self._evict()

    def _evict(self) -> None:
        # 粗估 4 字符 ≈ 1 token
        while sum(len(m["content"]) for m in self.messages) // 4 > self.budget_tokens \
                and len(self.messages) > 2:
            self.messages.pop(0)

 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

class ReActAgent:
    def __init__(self, llm, tools, max_steps=10):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.max_steps = max_steps

    def run(self, task: str) -> str:
        history = []
        for step in range(self.max_steps):
            prompt = self._build_prompt(task, history)
            response = self.llm.complete(prompt)
            parsed = self._parse(response)

            if parsed["type"] == "final_answer":
                return parsed["content"]

            tool = self.tools.get(parsed["tool"])
            if tool is None:
                obs = f"Unknown tool: {parsed['tool']}"
            else:
                result = tool(**parsed["args"])
                obs = result["result"] if result["ok"] else f"Error: {result['error']}"

            history.append({
                "thought": parsed.get("thought", ""),
                "action": f"{parsed['tool']}({parsed['args']})",
                "observation": obs,
            })
        return "Step budget exhausted"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

class VectorMemory:
    def __init__(self, embedder, store):
        self.embedder = embedder
        self.store = store  # 任意带 add/query 的向量后端

    def remember(self, text: str, metadata: dict) -> None:
        vec = self.embedder.embed(text)
        self.store.add(vec, text=text, **metadata)

    def recall(self, query: str, k: int = 5) -> list[str]:
        vec = self.embedder.embed(query)
        hits = self.store.query(vec, k=k)
        return [h["text"] for h in hits]

 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

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a city.",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "unit": {"type": "string", "enum": ["c", "f"]}
            },
            "required": ["city"]
        }
    }
}]


def run_turn(messages):
    resp = client.chat.completions.create(
        model="gpt-4o", messages=messages, tools=tools, tool_choice="auto"
    )
    msg = resp.choices[0].message
    if not msg.tool_calls:
        return msg.content

    messages.append(msg)
    for call in msg.tool_calls:
        args = json.loads(call.function.arguments)
        result = TOOL_REGISTRY[call.function.name](**args)
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": json.dumps(result),
        })
    # 第二轮：让模型把工具结果总结成自然语言
    return client.chat.completions.create(
        model="gpt-4o", messages=messages
    ).choices[0].message.content

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def self_refine(llm, task: str, max_iters: int = 3) -> str:
    output = llm.complete(f"Task: {task}\nAnswer:")
    for _ in range(max_iters):
        critique = llm.complete(
            f"Task: {task}\nAnswer: {output}\n"
            "List concrete problems. End with SCORE: 0-10."
        )
        score = parse_score(critique)
        if score >= 8:
            return output
        output = llm.complete(
            f"Task: {task}\nPrevious answer: {output}\n"
            f"Critique: {critique}\nImproved answer:"
        )
    return output

 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

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
import operator


class AgentState(TypedDict):
    messages: Annotated[Sequence[Any], operator.add]


def call_model(state: AgentState) -> dict:
    return {"messages": [llm.invoke(state["messages"])]}


def call_tools(state: AgentState) -> dict:
    last = state["messages"][-1]
    outputs = [TOOLS[c.name].invoke(c.args) for c in last.tool_calls]
    return {"messages": [ToolMessage(content=o, tool_call_id=c.id)
                          for o, c in zip(outputs, last.tool_calls)]}


def should_continue(state: AgentState) -> str:
    return "tools" if state["messages"][-1].tool_calls else END


graph = StateGraph(AgentState)
graph.add_node("model", call_model)
graph.add_node("tools", call_tools)
graph.set_entry_point("model")
graph.add_conditional_edges("model", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "model")
app = graph.compile()

1
2
3
4
5
6
7
8

@dataclass
class AgentMessage:
    sender: str
    recipient: str        # "*" 表示广播
    type: str             # "request" | "response" | "broadcast"
    content: dict
    correlation_id: str   # 把响应和请求关联起来
    timestamp: datetime

 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

@dataclass
class EvalCase:
    id: str
    input: str
    rubric: dict          # 评分维度：正确性、格式等
    must_use_tool: list   # 必须调用的工具
    must_not_call: list   # 禁止的动作（例如 dry-run 时禁止 send_email）
    timeout_s: int
    expected_cost_max: float


def evaluate(agent, cases: list[EvalCase]) -> dict:
    results = []
    for case in cases:
        with timeout(case.timeout_s):
            trace = agent.run(case.input)
        score = score_with_rubric(trace.output, case.rubric)
        results.append({
            "id": case.id,
            "score": score,
            "tools_used": trace.tools,
            "cost": trace.cost,
            "violated_constraints": check_constraints(trace, case),
        })
    return aggregate(results)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

class BudgetedAgent:
    def __init__(self, budget_usd: float):
        self.budget = budget_usd
        self.spent = 0.0

    def call(self, prompt, model="gpt-4o"):
        est = estimate_cost(prompt, model)
        if self.spent + est > self.budget:
            model = "gpt-4o-mini"  # 降级
        resp = client.complete(model=model, prompt=prompt)
        self.spent += actual_cost(resp)
        return resp

 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
69
70
71
72
73
74
75
76

import pandas as pd
import matplotlib.pyplot as plt
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import tool
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# 共享状态——生产环境里换成真正的存储
class State:
    raw: pd.DataFrame | None = None
    clean: pd.DataFrame | None = None
    charts: list[str] = []


@tool
def load_csv(path: str) -> str:
    """Load a CSV file into the analysis state."""
    State.raw = pd.read_csv(path)
    return f"Loaded {State.raw.shape[0]} rows x {State.raw.shape[1]} cols."


@tool
def clean() -> str:
    """Drop duplicates and impute missing values."""
    df = State.raw.drop_duplicates()
    num = df.select_dtypes("number").columns
    cat = df.select_dtypes("object").columns
    df[num] = df[num].fillna(df[num].mean())
    df[cat] = df[cat].fillna(df[cat].mode().iloc[0])
    State.clean = df
    return f"Cleaned. {len(df)} rows remain."


@tool
def describe() -> str:
    """Return summary statistics."""
    return State.clean.describe().to_string()


@tool
def plot(kind: str, x: str, y: str | None = None) -> str:
    """Create a plot. kind in {'hist','scatter','bar','box'}."""
    fig, ax = plt.subplots()
    df = State.clean
    if kind == "hist":
        ax.hist(df[x], bins=30)
    elif kind == "scatter" and y:
        ax.scatter(df[x], df[y])
    elif kind == "bar":
        df[x].value_counts().plot(kind="bar", ax=ax)
    elif kind == "box":
        df.boxplot(column=x, ax=ax)
    name = f"chart_{kind}_{x}.png"
    fig.savefig(name)
    plt.close(fig)
    State.charts.append(name)
    return f"Saved {name}"


tools = [load_csv, clean, describe, plot]
prompt = ChatPromptTemplate.from_messages([
    ("system",
     "You are a data analyst. Always: (1) load, (2) clean, "
     "(3) describe, (4) plot 2-3 informative charts, (5) summarise "
     "findings in 5 bullets. Never skip a step."),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

agent = create_openai_functions_agent(
    ChatOpenAI(model="gpt-4o", temperature=0), tools, prompt
)
executor = AgentExecutor(agent=agent, tools=tools,
                          max_iterations=10, verbose=True)

executor.invoke({"input": "Analyse sales_data.csv"})