1
from langsmith import traceable
2

3
@traceable(project_name="my-agent")
4
async def research_agent(query: str):
5
    # 每次调用自动追踪
6
    result = await llm.complete(query)
7
    return result

1
from opentelemetry import trace
2

3
tracer = trace.get_tracer(__name__)
4

5
@traceable
6
async def agent_with_spans(query: str):
7
    with tracer.start_as_current_span("route") as span:
8
        span.set_attribute("query_length", len(query))
9
        decision = await route(query)
10
        span.set_attribute("route_decision", decision)
11

12
    with tracer.start_as_current_span("execute"):
13
        result = await execute(decision)
14

15
    return result

1
import os
2
from langsmith import Client, traceable, trace
3
from langsmith.run_helpers import get_run_tree
4

5
# 初始化
6
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
7
os.environ["LANGCHAIN_TRACING_V2"] = "true"
8
os.environ["LANGCHAIN_PROJECT"] = "production-agent"
9

10
client = Client()
11

12
class TracedAgent:
13
    """带完整追踪的 Agent"""
14

15
    @traceable(name="agent.think", run_type="llm")
16
    async def think(self, query: str, context: str) -> str:
17
        """思考阶段追踪"""
18
        run = get_run_tree()
19
        if run:
20
            run.add_metadata({"query_tokens": count_tokens(query)})
21

22
        thought = await llm.complete(
23
            f"Context: {context}\nQuery: {query}\nThink step by step."
24
        )
25
        return thought
26

27
    @traceable(name="agent.act", run_type="tool")
28
    async def act(self, tool_name: str, tool_input: dict) -> str:
29
        """行动阶段追踪"""
30
        run = get_run_tree()
31
        if run:
32
            run.add_metadata({"tool": tool_name, "input": tool_input})
33

34
        result = await execute_tool(tool_name, tool_input)
35

36
        if run:
37
            run.add_metadata({"result_length": len(str(result))})
38
        return result
39

40
    @traceable(name="agent.run", run_type="chain")
41
    async def run(self, query: str) -> str:
42
        """完整 Agent 运行"""
43
        context = ""
44

45
        for step in range(MAX_STEPS):
46
            thought = await self.think(query, context)
47
            action = parse_action(thought)
48

49
            if action.type == "final_answer":
50
                return action.answer
51

52
            result = await self.act(action.tool, action.input)
53
            context += f"\nThought: {thought}\nAction: {action}\nObservation: {result}"
54

55
        return "达到最大步数"

1
from langsmith import Client
2

3
client = Client()
4

5
# 创建 Prompt 版本
6
prompt = client.push_prompt(
7
    "research-agent-system",
8
    object={
9
        "type": "chat",
10
        "messages": [
11
            {
12
                "role": "system",
13
                "content": "你是一个专业的研究助手，擅长搜索和分析信息。"
14
            }
15
        ],
16
    },
17
    metadata={"version": "2.1", "change": "增加了事实核查要求"},
18
)
19

20
# 拉取特定版本的 Prompt
21
prompt_v2 = client.pull_prompt("research-agent-system", metadata={"version": "2.0"})

1
from langfuse import Langfuse
2

3
langfuse = Langfuse()
4

5
@langfuse.trace()
6
def agent_step(tool_name: str, input_data: dict):
7
    # Langfuse 自动记录
8
    result = execute_tool(tool_name, input_data)
9
    langfuse.flush()
10
    return result

1
from langfuse import Langfuse
2
from langfuse.decorators import langfuse_context, observe
3

4
# 初始化（自托管地址）
5
langfuse = Langfuse(
6
    public_key="pk-xxx",
7
    secret_key="sk-xxx",
8
    host="http://localhost:3000",  # 自托管地址
9
)
10

11
class LangfuseAgent:
12
    """使用 Langfuse 追踪的 Agent"""
13

14
    @observe(name="agent.run", capture_input=True, capture_output=True)
15
    async def run(self, query: str, user_id: str = None) -> str:
16
        """Agent 主入口"""
17
        # 获取当前 trace
18
        trace = langfuse_context.get_current_trace()
19
        if trace:
20
            trace.update(
21
                user_id=user_id,
22
                metadata={"query_length": len(query)},
23
                tags=["production"],
24
            )
25

26
        # 思考
27
        thought = await self._think(query)
28

29
        # 行动
30
        result = await self._act(thought)
31

32
        # 记录评分（供后续评估用）
33
        langfuse_context.get_current_span().update(
34
            output=result,
35
            metadata={"steps_taken": thought.step_count},
36
        )
37

38
        return result
39

40
    @observe(name="agent.think")
41
    async def _think(self, query: str) -> dict:
42
        response = await llm.complete(query)
43
        return parse_thought(response)
44

45
    @observe(name="agent.act", kind="tool")
46
    async def _act(self, thought: dict) -> str:
47
        tool_name = thought["tool"]
48
        tool_input = thought["input"]
49

50
        result = await execute_tool(tool_name, tool_input)
51

52
        # 记录工具调用的详细信息
53
        span = langfuse_context.get_current_span()
54
        span.update(
55
            metadata={
56
                "tool_name": tool_name,
57
                "input_tokens": count_tokens(str(tool_input)),
58
                "output_tokens": count_tokens(result),
59
            }
60
        )
61

62
        return result

1
# 安装: pip install arize-phoenix
2
import phoenix as px
3
from phoenix.trace.langchain import LangChainInstrumentor
4

5
# 启动 Phoenix（本地）
6
session = px.launch_app()
7

8
# 自动追踪 LangChain 调用
9
LangChainInstrumentor().instrument()
10

11
# 现在所有 LangChain 调用都会被追踪
12
agent = create_react_agent(llm, tools, prompt)
13
result = agent.invoke({"input": "查询天气"})

1
from phoenix.experimental.evals import HallucinationEvaluator
2

3
# 使用 Phoenix 内置的幻觉评估器
4
evaluator = HallucinationEvaluator(model=llm)
5

6
# 评估 Agent 的回答
7
eval_results = []
8
for trace in session.get_traces():
9
    for span in trace.spans:
10
        if span.span_kind == "llm":
11
            result = evaluator.evaluate(
12
                input=span.input,
13
                output=span.output,
14
                reference=span.metadata.get("reference", ""),
15
            )
16
            eval_results.append(result)

1
import structlog
2
from datetime import datetime
3

4
# 配置 structlog
5
structlog.configure(
6
    processors=[
7
        structlog.contextvars.merge_contextvars,
8
        structlog.processors.add_log_level,
9
        structlog.processors.JSONRenderer(),
10
    ],
11
)
12

13
logger = structlog.get_logger()
14

15
class AgentLogger:
16
    """Agent 专用结构化日志"""
17

18
    def log_thought(self, trace_id: str, step: int, thought: str):
19
        logger.info(
20
            "agent.thought",
21
            trace_id=trace_id,
22
            step=step,
23
            phase="thought",
24
            content=thought[:200],  # 截断防止日志过大
25
            timestamp=datetime.now().isoformat(),
26
        )
27

28
    def log_action(self, trace_id: str, step: int, tool: str, params: dict):
29
        logger.info(
30
            "agent.action",
31
            trace_id=trace_id,
32
            step=step,
33
            phase="action",
34
            tool=tool,
35
            params=_sanitize_params(params),  # 脱敏
36
            timestamp=datetime.now().isoformat(),
37
        )
38

39
    def log_observation(self, trace_id: str, step: int, observation: str):
40
        logger.info(
41
            "agent.observation",
42
            trace_id=trace_id,
43
            step=step,
44
            phase="observation",
45
            content_length=len(observation),
46
            content_preview=observation[:100],
47
            timestamp=datetime.now().isoformat(),
48
        )
49

50
    def log_error(self, trace_id: str, step: int, error: Exception):
51
        logger.error(
52
            "agent.error",
53
            trace_id=trace_id,
54
            step=step,
55
            error_type=type(error).__name__,
56
            error_message=str(error),
57
            timestamp=datetime.now().isoformat(),
58
        )
59

60
    def log_token_usage(self, trace_id: str, model: str, prompt_tokens: int,
61
                        completion_tokens: int, cost_usd: float):
62
        logger.info(
63
            "agent.token_usage",
64
            trace_id=trace_id,
65
            model=model,
66
            prompt_tokens=prompt_tokens,
67
            completion_tokens=completion_tokens,
68
            total_tokens=prompt_tokens + completion_tokens,
69
            cost_usd=cost_usd,
70
            timestamp=datetime.now().isoformat(),
71
        )
72

73

74
def _sanitize_params(params: dict) -> dict:
75
    """参数脱敏：隐藏 API Key、密码等敏感信息"""
76
    sensitive_keys = {"api_key", "password", "token", "secret", "credential"}
77
    sanitized = {}
78
    for k, v in params.items():
79
        if any(s in k.lower() for s in sensitive_keys):
80
            sanitized[k] = "***REDACTED***"
81
        else:
82
            sanitized[k] = v
83
    return sanitized

1
{"event": "agent.thought", "trace_id": "t-001", "step": 1, "phase": "thought", "content": "用户询问北京天气，需要调用天气API获取实时数据", "timestamp": "2026-03-22T10:30:00"}
2
{"event": "agent.action", "trace_id": "t-001", "step": 1, "phase": "action", "tool": "get_weather", "params": {"city": "北京"}, "timestamp": "2026-03-22T10:30:01"}
3
{"event": "agent.observation", "trace_id": "t-001", "step": 1, "phase": "observation", "content_length": 85, "content_preview": "北京，晴，温度22°C，湿度45%，北风3级", "timestamp": "2026-03-22T10:30:02"}
4
{"event": "agent.token_usage", "trace_id": "t-001", "model": "gpt-4o", "prompt_tokens": 150, "completion_tokens": 80, "total_tokens": 230, "cost_usd": 0.00115, "timestamp": "2026-03-22T10:30:02"}

1
from opentelemetry import trace, metrics
2
from opentelemetry.sdk.trace import TracerProvider
3
from opentelemetry.sdk.trace.export import BatchSpanProcessor
4
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
5
from opentelemetry.sdk.metrics import MeterProvider
6
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
7
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
8

9
# 配置 Tracer
10
trace.set_tracer_provider(TracerProvider())
11
tracer = trace.get_tracer(__name__)
12
trace.get_tracer_provider().add_span_processor(
13
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317"))
14
)
15

16
# 配置 Meter
17
meter = metrics.get_meter(__name__)
18
metrics.set_meter_provider(MeterProvider(metric_readers=[
19
    PeriodicExportingMetricReader(
20
        OTLPMetricExporter(endpoint="http://localhost:4317")
21
    )
22
]))
23

24
# 定义指标
25
task_counter = meter.create_counter(
26
    name="agent.tasks.total",
27
    description="Total number of agent tasks",
28
    unit="1",
29
)
30
token_histogram = meter.create_histogram(
31
    name="agent.tokens.usage",
32
    description="Token usage per request",
33
    unit="tokens",
34
)
35
latency_histogram = meter.create_histogram(
36
    name="agent.request.latency",
37
    description="Request latency",
38
    unit="ms",
39
)
40

41
class OpenTelemetryAgent:
42
    """OpenTelemetry 追踪的 Agent"""
43

44
    async def run(self, query: str) -> str:
45
        with tracer.start_as_current_span("agent.run") as span:
46
            span.set_attribute("query", query[:100])
47
            span.set_attribute("query_length", len(query))
48

49
            start_time = time.time()
50
            try:
51
                result = await self._execute(query)
52
                task_counter.add(1, {"status": "success"})
53
                span.set_attribute("status", "success")
54
                return result
55
            except Exception as e:
56
                task_counter.add(1, {"status": "error", "error_type": type(e).__name__})
57
                span.set_attribute("status", "error")
58
                span.record_exception(e)
59
                raise
60
            finally:
61
                elapsed = (time.time() - start_time) * 1000
62
                latency_histogram.record(elapsed)
63

64
    async def _execute(self, query: str) -> str:
65
        with tracer.start_as_current_span("agent.think") as span:
66
            thought = await llm.complete(query)
67
            span.set_attribute("thought_length", len(thought))
68
            return thought
69

70
    def record_token_usage(self, model: str, prompt_tokens: int, completion_tokens: int):
71
        token_histogram.record(
72
            prompt_tokens + completion_tokens,
73
            {"model": model, "type": "total"},
74
        )

1
# 添加详细日志
2
import logging
3

4
logging.basicConfig(level=logging.DEBUG)
5

6
@traceable
7
async def debug_agent(query: str):
8
    logger.debug(f"输入: {query}")
9

10
    thought = await think(query)
11
    logger.debug(f"思考: {thought}")
12

13
    action = await act(thought)
14
    logger.debug(f"行动: {action}")
15

16
    return action

1
from langchain.debug import visualize_chains
2

3
# 开启可视化追踪
4
visualize_chains(agent_chain, visualization_dir="debug_output")

1
class TokenUsageTracker:
2
    """Token 用量追踪器"""
3

4
    def __init__(self):
5
        self.usage_by_model: dict[str, list[dict]] = {}
6
        self.usage_by_agent: dict[str, list[dict]] = {}
7

8
    def record(self, agent_name: str, model: str, usage: dict):
9
        record = {
10
            "timestamp": datetime.now().isoformat(),
11
            "prompt_tokens": usage.get("prompt_tokens", 0),
12
            "completion_tokens": usage.get("completion_tokens", 0),
13
            "total_tokens": usage.get("total_tokens", 0),
14
        }
15

16
        self.usage_by_model.setdefault(model, []).append(record)
17
        self.usage_by_agent.setdefault(agent_name, []).append(record)
18

19
    def get_summary(self, group_by: str = "model") -> dict:
20
        """获取用量汇总"""
21
        data = self.usage_by_model if group_by == "model" else self.usage_by_agent
22
        summary = {}
23
        for key, records in data.items():
24
            total_prompt = sum(r["prompt_tokens"] for r in records)
25
            total_completion = sum(r["completion_tokens"] for r in records)
26
            summary[key] = {
27
                "total_prompt_tokens": total_prompt,
28
                "total_completion_tokens": total_completion,
29
                "total_tokens": total_prompt + total_completion,
30
                "request_count": len(records),
31
                "avg_tokens_per_request": (total_prompt + total_completion) / len(records),
32
            }
33
        return summary

1
class CostDashboard:
2
    """实时成本仪表板"""
3

4
    def __init__(self):
5
        self.tracker = TokenUsageTracker()
6
        self.prices = MODEL_PRICING  # 参考前文的模型定价表
7

8
    def get_realtime_metrics(self) -> dict:
9
        summary = self.tracker.get_summary(group_by="model")
10
        total_cost = 0
11
        details = {}
12

13
        for model, stats in summary.items():
14
            if model in self.prices:
15
                cost = (
16
                    stats["total_prompt_tokens"] * self.prices[model]["input"] / 1_000_000
17
                    + stats["total_completion_tokens"] * self.prices[model]["output"] / 1_000_000
18
                )
19
            else:
20
                cost = 0
21
            total_cost += cost
22
            details[model] = {**stats, "cost_usd": cost}
23

24
        return {
25
            "total_cost_usd": total_cost,
26
            "by_model": details,
27
            "timestamp": datetime.now().isoformat(),
28
        }
29

30
    def check_budget(self, budget_usd: float) -> dict:
31
        metrics = self.get_realtime_metrics()
32
        remaining = budget_usd - metrics["total_cost_usd"]
33
        usage_pct = metrics["total_cost_usd"] / budget_usd
34

35
        alert = None
36
        if usage_pct > 0.9:
37
            alert = "CRITICAL: 已使用 90% 以上预算"
38
        elif usage_pct > 0.7:
39
            alert = "WARNING: 已使用 70% 以上预算"
40

41
        return {
42
            "budget_usd": budget_usd,
43
            "used_usd": metrics["total_cost_usd"],
44
            "remaining_usd": remaining,
45
            "usage_pct": usage_pct,
46
            "alert": alert,
47
        }

1
# Prometheus 指标
2
agent_tasks_total{status="completed", agent="research"} 1000
3
agent_tasks_total{status="failed", agent="research"} 5
4
agent_latency_seconds{agent="research"} 2.5
5
agent_cost_tokens{agent="research"} 500000

1
# alerting rules
2
groups:
3
  - name: agent
4
    rules:
5
      - alert: HighFailureRate
6
        expr: rate(agent_tasks_failed / agent_tasks_total) > 0.1
7
        for: 5m
8
        labels:
9
          severity: warning

1
# Grafana Dashboard 配置建议
2
DASHBOARD_PANELS = {
3
    "row_1": [
4
        {"title": "请求成功率", "type": "stat", "query": 'sum(rate(agent_tasks_total{status="success"}[5m])) / sum(rate(agent_tasks_total[5m]))'},
5
        {"title": "平均延迟 (ms)", "type": "gauge", "query": "histogram_quantile(0.5, agent_latency_seconds)"},
6
        {"title": "P99 延迟 (ms)", "type": "gauge", "query": "histogram_quantile(0.99, agent_latency_seconds)"},
7
        {"title": "每小时成本 ($)", "type": "stat", "query": 'sum(rate(agent_cost_usd[1h]))'},
8
    ],
9
    "row_2": [
10
        {"title": "Token 使用趋势", "type": "timeseries", "query": "sum by (model) (rate(agent_tokens_total[5m]))"},
11
        {"title": "工具调用成功率", "type": "bar_chart", "query": "sum by (tool) (rate(agent_tool_calls{status='success'}[5m]))"},
12
        {"title": "错误分布", "type": "pie_chart", "query": "sum by (error_type) (rate(agent_errors_total[5m]))"},
13
    ],
14
    "row_3": [
15
        {"title": "Agent 循环检测", "type": "stat", "query": 'sum(rate(agent_loop_detected_total[5m]))'},
16
        {"title": "上下文利用率", "type": "gauge", "query": "avg(agent_context_utilization)"},
17
        {"title": "缓存命中率", "type": "stat", "query": 'sum(rate(agent_cache_hits[5m])) / sum(rate(agent_cache_requests[5m]))'},
18
    ],
19
}