#!/usr/bin/env python3 """Real local LangChain tool-loop exercise with scripted model decisions. Python 3.11+; pip install langchain==1.4.6 langgraph==1.2.16 in an isolated venv. Run: python fde-langchain-agent-lab.py test python fde-langchain-agent-lab.py supported Cases: supported, foreign-source, authority-override, unknown-tool, spin. Synthetic records only. No inference provider, API key, sending or network tool. """ import argparse import json import os import unittest from unittest.mock import patch # The installed framework may otherwise inherit tracing from a learner's shell. # Keep this explicitly local lab out of remote tracing even when it is configured. os.environ["LANGSMITH_TRACING"] = "false" os.environ["LANGCHAIN_TRACING_V2"] = "false" from langchain.agents import create_agent from langchain.tools import tool from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import AIMessage, ToolMessage from langchain_core.outputs import ChatGeneration, ChatResult from langgraph.errors import GraphRecursionError from pydantic import BaseModel, ConfigDict, Field CASES = ["supported", "foreign-source", "authority-override", "unknown-tool", "spin"] OTHER_TENANT_TEXT = "Synthetic south-only replacement procedure." class LookupInput(BaseModel): model_config = ConfigDict(extra="forbid") procedure_id: str = Field(pattern=r"^P[0-9]{1,3}$") class ScriptedDecisionModel(BaseChatModel): """A deterministic test adapter, NOT an inference model or a quality judge. LangChain still performs the actual model/tool/model orchestration. This adapter supplies hostile and useful tool proposals so boundary cases repeat. """ scenario: str = "supported" @property def _llm_type(self): return "lockedin-scripted-training-model" def bind_tools(self, tools, **kwargs): # A real provider adapter would serialize these schemas for the model. # This scripted adapter deliberately emits the case's predefined proposal. return self def _generate(self, messages, stop=None, run_manager=None, **kwargs): last = messages[-1] if isinstance(last, ToolMessage) and self.scenario != "spin": if last.status == "error": response = AIMessage(content="Withheld: the proposed tool call was refused.") else: result = json.loads(last.content) response = AIMessage(content=("Operator draft: " + result["excerpt"]) if result["status"] == "found" else "Withheld: source unavailable.") else: name = "send_message" if self.scenario == "unknown-tool" else "lookup_procedure" args = {"procedure_id": "P9" if self.scenario == "foreign-source" else "P1"} if self.scenario == "authority-override": args["tenant"] = "south" count = sum(isinstance(message, AIMessage) for message in messages) response = AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": f"call-{count}"}]) return ChatResult(generations=[ChatGeneration(message=response)]) def build_agent(scenario, audit): # Trusted synthetic host scope, never supplied by the scripted model. # A deployed service must derive this from authenticated caller context. trusted_tenant = "north" sources = {"P1": ("north", "Inspect the disconnected training unit."), "P9": ("south", OTHER_TENANT_TEXT)} @tool(args_schema=LookupInput) def lookup_procedure(procedure_id: str) -> str: """Read one currently approved procedure by ID. No writes or arbitrary URLs.""" source = sources.get(procedure_id) admitted = bool(source and source[0] == trusted_tenant) audit.append({"tool": "lookup_procedure", "admitted": admitted}) if not admitted: return json.dumps({"status": "unavailable"}) return json.dumps({"status": "found", "source_id": procedure_id, "version": 1, "excerpt": source[1]}) return create_agent(model=ScriptedDecisionModel(scenario=scenario), tools=[lookup_procedure], system_prompt="Draft from one approved procedure or explain why you cannot.") def run_case(scenario): if scenario not in CASES: raise ValueError("unknown_case") audit = [] agent = build_agent(scenario, audit) try: result = agent.invoke({"messages": [{"role": "user", "content": "Prepare a synthetic operator draft."}]}, config={"recursion_limit": 8}) messages = result["messages"] return {"status": "finished", "answer": messages[-1].content, "audit": audit, "trajectory": ["tool-result" if isinstance(m, ToolMessage) else "tool-proposal" if isinstance(m, AIMessage) and m.tool_calls else "answer" if isinstance(m, AIMessage) else "input" for m in messages]} except GraphRecursionError: return {"status": "incomplete", "reason": "graph_step_limit", "audit": audit} class AgentLoopTests(unittest.TestCase): def setUp(self): self.network = patch("socket.socket.connect", side_effect=AssertionError("Lab must stay offline")) self.network.start() def tearDown(self): self.network.stop() def test_authorized_task_uses_real_tool_loop(self): run = run_case("supported") self.assertEqual(run["status"], "finished") self.assertIn("Inspect the disconnected", run["answer"]) self.assertEqual(run["trajectory"], ["input", "tool-proposal", "tool-result", "answer"]) self.assertEqual(run["audit"], [{"tool": "lookup_procedure", "admitted": True}]) def test_foreign_source_is_withheld_before_context(self): run = run_case("foreign-source") self.assertIn("Withheld", run["answer"]) self.assertFalse(run["audit"][0]["admitted"]) self.assertNotIn(OTHER_TENANT_TEXT, json.dumps(run)) def test_model_cannot_add_tenant_authority(self): run = run_case("authority-override") self.assertIn("refused", run["answer"]) self.assertEqual(run["audit"], []) # Schema refusal before business lookup. def test_ungranted_tool_is_not_dispatched(self): run = run_case("unknown-tool") self.assertIn("refused", run["answer"]) self.assertEqual(run["audit"], []) def test_repeating_agent_stops_incomplete_at_graph_limit(self): run = run_case("spin") self.assertEqual(run["status"], "incomplete") self.assertEqual(run["reason"], "graph_step_limit") self.assertGreater(len(run["audit"]), 0) self.assertLess(len(run["audit"]), 8) self.assertNotIn("answer", run) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("case", choices=["test", *CASES]) args = parser.parse_args() if args.case == "test": suite = unittest.defaultTestLoader.loadTestsFromTestCase(AgentLoopTests) return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1 result = run_case(args.case) print(json.dumps(result, indent=2)) print("Real LangChain orchestration; model decisions are scripted. No message was sent.") return 0 if result["status"] == "finished" else 2 if __name__ == "__main__": raise SystemExit(main())