用 Omnigent 构建安全的多智能体财务工作流

MarkTechPost#Agent#多智能体

Editor brief · 编辑速览

教程展示如何集成实时数据、实现多层级智能体协作和成本治理策略，实现端到端财务分析工作流。

import os, sys, subprocess, textwrap, pathlib, getpass def sh(cmd, **kw): """Run a command, and on failure show the ACTUAL error, not just a code.""" print("$", " ".join(map(str, cmd))) p = subprocess.run(cmd, text=True, capture_output=True, **kw) if p.returncode != 0: print(p.stdout or "", p.stderr or "", sep="\n") raise RuntimeError(f"Command failed ({p.returncode}): {' '.join(map(str, cmd))}") return p WORKDIR = pathlib.Path("/content/omnigent_tutorial") WORKDIR.mkdir(parents=True, exist_ok=True) VENV = WORKDIR / ".venv" subprocess.run([sys.executable, "-m", "pip", "install", "-q", "uv"], check=True) if not (VENV / "bin" / "python").exists(): sh(["uv", "venv", "--python", "3.12", str(VENV)]) PY = str(VENV / "bin" / "python") sh(["uv", "pip", "install", "--python", PY, "-q", "omnigent", "requests"]) OMNI = str(VENV / "bin" / "omnigent") print("\n✅", subprocess.run([OMNI, "--version"], capture_output=True, text=True).stdout.strip())

(WORKDIR / "agent_tools.py").write_text(textwrap.dedent(''' """Local tools exposed to the Omnigent agents in this tutorial.""" import requests def get_exchange_rate(base_currency: str, target_currency: str) -> dict: """Look up the latest FX rate between two ISO-4217 currency codes.""" r = requests.get( "https://api.frankfurter.app/latest", params={"from": base_currency.upper(), "to": target_currency.upper()}, timeout=10, ) r.raise_for_status() data = r.json() return { "base": base_currency.upper(), "target": target_currency.upper(), "rate": data["rates"][target_currency.upper()], "date": data["date"], } def word_count(text: str) -> int: """Count the words in a piece of text.""" return len(text.split()) '''))

(WORKDIR / "fx_research_lead.yaml").write_text(textwrap.dedent(''' name: fx_research_lead prompt: | You are a financial research lead. For any question about currency movements: call get_exchange_rate to fetch the live rate, then hand your draft summary to the text_auditor sub-agent for a clarity and length check before giving your final answer to the user. executor: harness: claude-sdk tools: get_exchange_rate: type: function callable: agent_tools.get_exchange_rate text_auditor: type: agent prompt: | You audit short pieces of financial writing. Call word_count to report its length, flag any unexplained jargon, and suggest one concrete clarity improvement. tools: word_count: type: function callable: agent_tools.word_count policies: cap_calls: type: function handler: omnigent.policies.builtins.safety.max_tool_calls_per_session factory_params: limit: 20 budget: type: function handler: omnigent.policies.builtins.cost.cost_budget factory_params: max_cost_usd: 1.00 '''))

env["PYTHONPATH"] = str(WORKDIR) question = ( "What is the current USD to EUR exchange rate? Give me a two-sentence " "summary I could paste into a client note." ) result = subprocess.run( [OMNI, "run", str(WORKDIR / "fx_research_lead.yaml"), "-p", question, "--no-session"], cwd=WORKDIR, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=300, ) print("\n" + "=" * 70) print(result.stdout.strip() or "(no stdout)") if result.returncode != 0 or "error" in result.stdout.lower(): print("-" * 70) print("stderr:", result.stderr[-2000:]) print(f"\nDebug: check ~/.omnigent/logs/runner/ , or rerun with:\n" f" !{OMNI} --debug --log-to-stderr run {WORKDIR/'fx_research_lead.yaml'} -p \"...\" --no-session") print("=" * 70) print(f""" Next steps: • Explore the CLI: !{OMNI} run --help • Bundled demo agents: !{OMNI} polly -p "review this repo" --no-session !{OMNI} debby -p "brainstorm 3 names for a coffee shop" --no-session • YAML schema: https://github.com/omnigent-ai/omnigent/blob/main/docs/AGENT_YAML_SPEC.md • Policies: https://github.com/omnigent-ai/omnigent/blob/main/docs/POLICIES.md """)

完整中文译文